diff --git a/api4/api.go b/api4/api.go index 7d193373c6..0e8ba10e26 100644 --- a/api4/api.go +++ b/api4/api.go @@ -21,6 +21,8 @@ type Routes struct { Users *mux.Router // 'api/v4/users' User *mux.Router // 'api/v4/users/{user_id:[A-Za-z0-9]+}' + UserThreads *mux.Router // 'api/v4/users/{user_id:[A-Za-z0-9]+}/threads' + UserThread *mux.Router // 'api/v4/users/{user_id:[A-Za-z0-9]+}/threads/{thread_id:[A-Za-z0-9]+}' UserByUsername *mux.Router // 'api/v4/users/username/{username:[A-Za-z0-9\\_\\-\\.]+}' UserByEmail *mux.Router // 'api/v4/users/email/{email:.+}' @@ -141,6 +143,8 @@ func Init(configservice configservice.ConfigService, globalOptionsFunc app.AppOp api.BaseRoutes.Users = api.BaseRoutes.ApiRoot.PathPrefix("/users").Subrouter() api.BaseRoutes.User = api.BaseRoutes.ApiRoot.PathPrefix("/users/{user_id:[A-Za-z0-9]+}").Subrouter() + api.BaseRoutes.UserThreads = api.BaseRoutes.ApiRoot.PathPrefix("/users/{user_id:[A-Za-z0-9]+}/threads").Subrouter() + api.BaseRoutes.UserThread = api.BaseRoutes.ApiRoot.PathPrefix("/users/{user_id:[A-Za-z0-9]+}/threads/{thread_id:[A-Za-z0-9]+}").Subrouter() api.BaseRoutes.UserByUsername = api.BaseRoutes.Users.PathPrefix("/username/{username:[A-Za-z0-9\\_\\-\\.]+}").Subrouter() api.BaseRoutes.UserByEmail = api.BaseRoutes.Users.PathPrefix("/email/{email:.+}").Subrouter() diff --git a/api4/user.go b/api4/user.go index aea417ff4f..219c3a3b62 100644 --- a/api4/user.go +++ b/api4/user.go @@ -91,6 +91,13 @@ func (api *API) InitUser() { api.BaseRoutes.Users.Handle("/migrate_auth/saml", api.ApiSessionRequired(migrateAuthToSaml)).Methods("POST") api.BaseRoutes.User.Handle("/uploads", api.ApiSessionRequired(getUploadsForUser)).Methods("GET") + + api.BaseRoutes.UserThreads.Handle("", api.ApiSessionRequired(getThreadsForUser)).Methods("GET") + api.BaseRoutes.UserThreads.Handle("/read/{timestamp:[0-9]+}", api.ApiSessionRequired(updateReadStateAllThreadsByUser)).Methods("PUT") + + api.BaseRoutes.UserThread.Handle("/following", api.ApiSessionRequired(followThreadByUser)).Methods("PUT") + api.BaseRoutes.UserThread.Handle("/following", api.ApiSessionRequired(unfollowThreadByUser)).Methods("DELETE") + api.BaseRoutes.UserThread.Handle("/read/{timestamp:[0-9]+}", api.ApiSessionRequired(updateReadStateThreadByUser)).Methods("PUT") } func createUser(c *Context, w http.ResponseWriter, r *http.Request) { @@ -2806,3 +2813,173 @@ func migrateAuthToSaml(c *Context, w http.ResponseWriter, r *http.Request) { auditRec.Success() ReturnStatusOK(w) } + +func getThreadsForUser(c *Context, w http.ResponseWriter, r *http.Request) { + c.RequireUserId() + if c.Err != nil { + return + } + + if !c.App.SessionHasPermissionToUser(*c.App.Session(), c.Params.UserId) { + c.SetPermissionError(model.PERMISSION_EDIT_OTHER_USERS) + return + } + + options := model.GetUserThreadsOpts{ + Since: 0, + Page: 0, + PageSize: 30, + Extended: false, + Deleted: false, + } + + sinceString := r.URL.Query().Get("since") + if len(sinceString) > 0 { + since, parseError := strconv.ParseUint(sinceString, 10, 64) + if parseError != nil { + c.SetInvalidParam("since") + return + } + options.Since = since + } + + pageString := r.URL.Query().Get("page") + if len(pageString) > 0 { + page, parseError := strconv.ParseUint(pageString, 10, 64) + if parseError != nil { + c.SetInvalidParam("page") + return + } + options.Page = page + } + + pageSizeString := r.URL.Query().Get("pageSize") + if len(pageString) > 0 { + pageSize, parseError := strconv.ParseUint(pageSizeString, 10, 64) + if parseError != nil { + c.SetInvalidParam("pageSize") + return + } + options.PageSize = pageSize + } + + deletedStr := r.URL.Query().Get("deleted") + extendedStr := r.URL.Query().Get("extended") + + options.Deleted, _ = strconv.ParseBool(deletedStr) + options.Extended, _ = strconv.ParseBool(extendedStr) + + threads, err := c.App.GetThreadsForUser(c.Params.UserId, options) + if err != nil { + c.Err = err + return + } + + w.Write([]byte(threads.ToJson())) +} + +func updateReadStateThreadByUser(c *Context, w http.ResponseWriter, r *http.Request) { + c.RequireUserId().RequireThreadId().RequireTimestamp() + if c.Err != nil { + return + } + + auditRec := c.MakeAuditRecord("updateReadStateThreadByUser", audit.Fail) + defer c.LogAuditRec(auditRec) + auditRec.AddMeta("user_id", c.Params.UserId) + auditRec.AddMeta("thread_id", c.Params.ThreadId) + auditRec.AddMeta("timestamp", c.Params.Timestamp) + if !c.App.SessionHasPermissionToUser(*c.App.Session(), c.Params.UserId) { + c.SetPermissionError(model.PERMISSION_EDIT_OTHER_USERS) + return + } + + err := c.App.UpdateThreadReadForUser(c.Params.UserId, c.Params.ThreadId, c.Params.Timestamp) + if err != nil { + c.Err = err + return + } + + ReturnStatusOK(w) + + auditRec.Success() +} + +func unfollowThreadByUser(c *Context, w http.ResponseWriter, r *http.Request) { + c.RequireUserId().RequireThreadId() + if c.Err != nil { + return + } + + auditRec := c.MakeAuditRecord("unfollowThreadByUser", audit.Fail) + defer c.LogAuditRec(auditRec) + auditRec.AddMeta("user_id", c.Params.UserId) + auditRec.AddMeta("thread_id", c.Params.ThreadId) + + if !c.App.SessionHasPermissionToUser(*c.App.Session(), c.Params.UserId) { + c.SetPermissionError(model.PERMISSION_EDIT_OTHER_USERS) + return + } + + err := c.App.UpdateThreadFollowForUser(c.Params.UserId, c.Params.ThreadId, false) + if err != nil { + c.Err = err + return + } + + ReturnStatusOK(w) + + auditRec.Success() +} + +func followThreadByUser(c *Context, w http.ResponseWriter, r *http.Request) { + c.RequireUserId().RequireThreadId() + if c.Err != nil { + return + } + + auditRec := c.MakeAuditRecord("followThreadByUser", audit.Fail) + defer c.LogAuditRec(auditRec) + auditRec.AddMeta("user_id", c.Params.UserId) + auditRec.AddMeta("thread_id", c.Params.ThreadId) + + if !c.App.SessionHasPermissionToUser(*c.App.Session(), c.Params.UserId) { + c.SetPermissionError(model.PERMISSION_EDIT_OTHER_USERS) + return + } + + err := c.App.UpdateThreadFollowForUser(c.Params.UserId, c.Params.ThreadId, true) + if err != nil { + c.Err = err + return + } + + ReturnStatusOK(w) + auditRec.Success() +} + +func updateReadStateAllThreadsByUser(c *Context, w http.ResponseWriter, r *http.Request) { + c.RequireUserId().RequireTimestamp() + if c.Err != nil { + return + } + + auditRec := c.MakeAuditRecord("updateReadStateAllThreadsByUser", audit.Fail) + defer c.LogAuditRec(auditRec) + auditRec.AddMeta("user_id", c.Params.UserId) + auditRec.AddMeta("timestamp", c.Params.Timestamp) + + if !c.App.SessionHasPermissionToUser(*c.App.Session(), c.Params.UserId) { + c.SetPermissionError(model.PERMISSION_EDIT_OTHER_USERS) + return + } + + err := c.App.UpdateThreadsReadForUser(c.Params.UserId, c.Params.Timestamp) + if err != nil { + c.Err = err + return + } + + ReturnStatusOK(w) + auditRec.Success() +} diff --git a/api4/user_test.go b/api4/user_test.go index 362a14f451..65a3b0deb4 100644 --- a/api4/user_test.go +++ b/api4/user_test.go @@ -5255,3 +5255,308 @@ func TestUpdatePassword(t *testing.T) { CheckOKStatus(t, res) }) } + +func TestGetThreadsForUser(t *testing.T) { + th := Setup(t).InitBasic() + defer th.TearDown() + + t.Run("empty", func(t *testing.T) { + Client := th.Client + + _, resp := Client.CreatePost(&model.Post{ChannelId: th.BasicChannel.Id, Message: "testMsg"}) + CheckNoError(t, resp) + CheckCreatedStatus(t, resp) + + defer th.App.Srv().Store.Post().PermanentDeleteByUser(th.BasicUser.Id) + + uss, resp := th.Client.GetUserThreads(th.BasicUser.Id, model.GetUserThreadsOpts{ + Page: 0, + PageSize: 30, + }) + require.Nil(t, resp.Error) + require.Len(t, uss.Threads, 0) + }) + + t.Run("no params, 1 thread", func(t *testing.T) { + Client := th.Client + + rpost, resp := Client.CreatePost(&model.Post{ChannelId: th.BasicChannel.Id, Message: "testMsg"}) + CheckNoError(t, resp) + CheckCreatedStatus(t, resp) + _, resp2 := Client.CreatePost(&model.Post{ChannelId: th.BasicChannel.Id, Message: "testReply", RootId: rpost.Id}) + CheckNoError(t, resp2) + CheckCreatedStatus(t, resp2) + + defer th.App.Srv().Store.Post().PermanentDeleteByUser(th.BasicUser.Id) + + uss, resp := th.Client.GetUserThreads(th.BasicUser.Id, model.GetUserThreadsOpts{ + Page: 0, + PageSize: 30, + }) + require.Nil(t, resp.Error) + require.Len(t, uss.Threads, 1) + require.Equal(t, uss.Threads[0].PostId, rpost.Id) + require.Equal(t, uss.Threads[0].ReplyCount, int64(1)) + }) + + t.Run("extended, 1 thread", func(t *testing.T) { + Client := th.Client + + rpost, resp := Client.CreatePost(&model.Post{ChannelId: th.BasicChannel.Id, Message: "testMsg"}) + CheckNoError(t, resp) + CheckCreatedStatus(t, resp) + _, resp2 := Client.CreatePost(&model.Post{ChannelId: th.BasicChannel.Id, Message: "testReply", RootId: rpost.Id}) + CheckNoError(t, resp2) + CheckCreatedStatus(t, resp2) + + defer th.App.Srv().Store.Post().PermanentDeleteByUser(th.BasicUser.Id) + + uss, resp := th.Client.GetUserThreads(th.BasicUser.Id, model.GetUserThreadsOpts{ + Page: 0, + PageSize: 30, + Extended: true, + }) + require.Nil(t, resp.Error) + require.Len(t, uss.Threads, 1) + require.Equal(t, uss.Threads[0].PostId, rpost.Id) + require.Equal(t, uss.Threads[0].ReplyCount, int64(1)) + require.Equal(t, uss.Threads[0].Participants[0].Id, th.BasicUser.Id) + }) + + t.Run("deleted, 1 thread", func(t *testing.T) { + Client := th.Client + + rpost, resp := Client.CreatePost(&model.Post{ChannelId: th.BasicChannel.Id, Message: "testMsg"}) + CheckNoError(t, resp) + CheckCreatedStatus(t, resp) + _, resp2 := Client.CreatePost(&model.Post{ChannelId: th.BasicChannel.Id, Message: "testReply", RootId: rpost.Id}) + CheckNoError(t, resp2) + CheckCreatedStatus(t, resp2) + + defer th.App.Srv().Store.Post().PermanentDeleteByUser(th.BasicUser.Id) + + uss, resp := th.Client.GetUserThreads(th.BasicUser.Id, model.GetUserThreadsOpts{ + Page: 0, + PageSize: 30, + Deleted: false, + }) + require.Nil(t, resp.Error) + require.Len(t, uss.Threads, 1) + require.Equal(t, uss.Threads[0].PostId, rpost.Id) + require.Equal(t, uss.Threads[0].ReplyCount, int64(1)) + require.Equal(t, uss.Threads[0].Participants[0].Id, th.BasicUser.Id) + + res, resp2 := th.Client.DeletePost(rpost.Id) + require.True(t, res) + require.Nil(t, resp2.Error) + + uss, resp = th.Client.GetUserThreads(th.BasicUser.Id, model.GetUserThreadsOpts{ + Page: 0, + PageSize: 30, + Deleted: false, + }) + require.Nil(t, resp.Error) + require.Len(t, uss.Threads, 0) + + uss, resp = th.Client.GetUserThreads(th.BasicUser.Id, model.GetUserThreadsOpts{ + Page: 0, + PageSize: 30, + Deleted: true, + }) + require.Nil(t, resp.Error) + require.Len(t, uss.Threads, 1) + require.Greater(t, uss.Threads[0].Post.DeleteAt, int64(0)) + + }) + + t.Run("paged, 30 threads", func(t *testing.T) { + Client := th.Client + + var rootIds []*model.Post + for i := 0; i < 30; i++ { + time.Sleep(1) + rpost, resp := Client.CreatePost(&model.Post{ChannelId: th.BasicChannel.Id, Message: "testMsg"}) + CheckNoError(t, resp) + CheckCreatedStatus(t, resp) + rootIds = append(rootIds, rpost) + time.Sleep(1) + _, resp2 := Client.CreatePost(&model.Post{ChannelId: th.BasicChannel.Id, Message: "testReply", RootId: rpost.Id}) + CheckNoError(t, resp2) + CheckCreatedStatus(t, resp2) + } + + defer th.App.Srv().Store.Post().PermanentDeleteByUser(th.BasicUser.Id) + + uss, resp := th.Client.GetUserThreads(th.BasicUser.Id, model.GetUserThreadsOpts{ + Page: 0, + PageSize: 30, + Deleted: false, + }) + require.Nil(t, resp.Error) + require.Len(t, uss.Threads, 30) + require.Len(t, rootIds, 30) + require.Equal(t, uss.Threads[0].PostId, rootIds[29].Id) + require.Equal(t, uss.Threads[0].ReplyCount, int64(1)) + require.Equal(t, uss.Threads[0].Participants[0].Id, th.BasicUser.Id) + }) +} + +func TestFollowThreads(t *testing.T) { + th := Setup(t).InitBasic() + defer th.TearDown() + + t.Run("1 thread", func(t *testing.T) { + Client := th.Client + + rpost, resp := Client.CreatePost(&model.Post{ChannelId: th.BasicChannel.Id, Message: "testMsg"}) + CheckNoError(t, resp) + CheckCreatedStatus(t, resp) + _, resp2 := Client.CreatePost(&model.Post{ChannelId: th.BasicChannel.Id, Message: "testReply", RootId: rpost.Id}) + CheckNoError(t, resp2) + CheckCreatedStatus(t, resp2) + + defer th.App.Srv().Store.Post().PermanentDeleteByUser(th.BasicUser.Id) + var uss *model.Threads + uss, resp = th.Client.GetUserThreads(th.BasicUser.Id, model.GetUserThreadsOpts{ + Page: 0, + PageSize: 30, + Deleted: false, + }) + CheckNoError(t, resp) + require.Len(t, uss.Threads, 1) + + resp = th.Client.UpdateThreadFollowForUser(th.BasicUser.Id, rpost.Id, false) + CheckNoError(t, resp) + CheckOKStatus(t, resp) + + uss, resp = th.Client.GetUserThreads(th.BasicUser.Id, model.GetUserThreadsOpts{ + Page: 0, + PageSize: 30, + Deleted: false, + }) + CheckNoError(t, resp) + require.Len(t, uss.Threads, 0) + + resp = th.Client.UpdateThreadFollowForUser(th.BasicUser.Id, rpost.Id, true) + CheckNoError(t, resp) + CheckOKStatus(t, resp) + + uss, resp = th.Client.GetUserThreads(th.BasicUser.Id, model.GetUserThreadsOpts{ + Page: 0, + PageSize: 30, + Deleted: false, + }) + CheckNoError(t, resp) + require.Len(t, uss.Threads, 1) + + }) +} + +func TestReadThreads(t *testing.T) { + th := Setup(t).InitBasic() + defer th.TearDown() + + t.Run("all threads", func(t *testing.T) { + Client := th.Client + + rpost, resp := Client.CreatePost(&model.Post{ChannelId: th.BasicChannel.Id, Message: "testMsg"}) + CheckNoError(t, resp) + CheckCreatedStatus(t, resp) + rpost2, resp2 := Client.CreatePost(&model.Post{ChannelId: th.BasicChannel.Id, Message: "testReply", RootId: rpost.Id}) + CheckNoError(t, resp2) + CheckCreatedStatus(t, resp2) + defer th.App.Srv().Store.Post().PermanentDeleteByUser(th.BasicUser.Id) + + var uss, uss2, uss3 *model.Threads + uss, resp = th.Client.GetUserThreads(th.BasicUser.Id, model.GetUserThreadsOpts{ + Page: 0, + PageSize: 30, + Deleted: false, + }) + CheckNoError(t, resp) + require.Len(t, uss.Threads, 1) + + time.Sleep(1) + resp = th.Client.UpdateThreadsReadForUser(th.BasicUser.Id, model.GetMillis()) + CheckNoError(t, resp) + CheckOKStatus(t, resp) + + uss2, resp = th.Client.GetUserThreads(th.BasicUser.Id, model.GetUserThreadsOpts{ + Page: 0, + PageSize: 30, + Deleted: false, + }) + CheckNoError(t, resp) + require.Len(t, uss2.Threads, 1) + require.Greater(t, uss2.Threads[0].LastViewedAt, uss.Threads[0].LastViewedAt) + + resp = th.Client.UpdateThreadsReadForUser(th.BasicUser.Id, rpost2.UpdateAt) + CheckNoError(t, resp) + CheckOKStatus(t, resp) + + uss3, resp = th.Client.GetUserThreads(th.BasicUser.Id, model.GetUserThreadsOpts{ + Page: 0, + PageSize: 30, + Deleted: false, + }) + CheckNoError(t, resp) + require.Len(t, uss3.Threads, 1) + require.Equal(t, uss3.Threads[0].LastViewedAt, rpost2.UpdateAt) + }) + + t.Run("1 thread", func(t *testing.T) { + Client := th.Client + + rpost, resp := Client.CreatePost(&model.Post{ChannelId: th.BasicChannel.Id, Message: "testMsg"}) + CheckNoError(t, resp) + CheckCreatedStatus(t, resp) + _, resp2 := Client.CreatePost(&model.Post{ChannelId: th.BasicChannel.Id, Message: "testReply", RootId: rpost.Id}) + CheckNoError(t, resp2) + CheckCreatedStatus(t, resp2) + + rrpost, rresp := Client.CreatePost(&model.Post{ChannelId: th.BasicChannel.Id, Message: "testMsg"}) + CheckNoError(t, rresp) + CheckCreatedStatus(t, rresp) + _, rresp2 := Client.CreatePost(&model.Post{ChannelId: th.BasicChannel.Id, Message: "testReply", RootId: rrpost.Id}) + CheckNoError(t, rresp2) + CheckCreatedStatus(t, rresp2) + + defer th.App.Srv().Store.Post().PermanentDeleteByUser(th.BasicUser.Id) + + var uss, uss2, uss3 *model.Threads + uss, resp = th.Client.GetUserThreads(th.BasicUser.Id, model.GetUserThreadsOpts{ + Page: 0, + PageSize: 30, + Deleted: false, + }) + CheckNoError(t, resp) + require.Len(t, uss.Threads, 2) + + resp = th.Client.UpdateThreadReadForUser(th.BasicUser.Id, rrpost.Id, model.GetMillis()) + CheckNoError(t, resp) + CheckOKStatus(t, resp) + + uss2, resp = th.Client.GetUserThreads(th.BasicUser.Id, model.GetUserThreadsOpts{ + Page: 0, + PageSize: 30, + Deleted: false, + }) + CheckNoError(t, resp) + require.Len(t, uss2.Threads, 2) + require.Greater(t, uss2.Threads[1].LastViewedAt, uss.Threads[1].LastViewedAt) + + timestamp := model.GetMillis() + resp = th.Client.UpdateThreadReadForUser(th.BasicUser.Id, rrpost.Id, timestamp) + CheckNoError(t, resp) + CheckOKStatus(t, resp) + + uss3, resp = th.Client.GetUserThreads(th.BasicUser.Id, model.GetUserThreadsOpts{ + Page: 0, + PageSize: 30, + Deleted: false, + }) + CheckNoError(t, resp) + require.Len(t, uss3.Threads, 2) + require.Equal(t, uss3.Threads[1].LastViewedAt, timestamp) + }) +} diff --git a/app/app_iface.go b/app/app_iface.go index 73f53f625d..72351b70ba 100644 --- a/app/app_iface.go +++ b/app/app_iface.go @@ -687,6 +687,7 @@ type AppIface interface { GetTeamsUnreadForUser(excludeTeamId string, userId string) ([]*model.TeamUnread, *model.AppError) GetTermsOfService(id string) (*model.TermsOfService, *model.AppError) GetThreadMembershipsForUser(userId string) ([]*model.ThreadMembership, error) + GetThreadsForUser(userId string, options model.GetUserThreadsOpts) (*model.Threads, *model.AppError) GetUploadSession(uploadId string) (*model.UploadSession, *model.AppError) GetUploadSessionsForUser(userId string) ([]*model.UploadSession, *model.AppError) GetUser(userId string) (*model.User, *model.AppError) @@ -993,6 +994,9 @@ type AppIface interface { UpdateTeamMemberSchemeRoles(teamId string, userId string, isSchemeGuest bool, isSchemeUser bool, isSchemeAdmin bool) (*model.TeamMember, *model.AppError) UpdateTeamPrivacy(teamId string, teamType string, allowOpenInvite bool) *model.AppError UpdateTeamScheme(team *model.Team) (*model.Team, *model.AppError) + UpdateThreadFollowForUser(userId, threadId string, state bool) *model.AppError + UpdateThreadReadForUser(userId, threadId string, timestamp int64) *model.AppError + UpdateThreadsReadForUser(userId string, timestamp int64) *model.AppError UpdateUser(user *model.User, sendNotifications bool) (*model.User, *model.AppError) UpdateUserActive(userId string, active bool) *model.AppError UpdateUserAsUser(user *model.User, asAdmin bool) (*model.User, *model.AppError) diff --git a/app/notification.go b/app/notification.go index fd5441cda8..fb1607d925 100644 --- a/app/notification.go +++ b/app/notification.go @@ -167,7 +167,7 @@ func (a *App) SendNotifications(post *model.Post, team *model.Team, channel *mod go func(userId string) { defer close(mac) if *a.Config().ServiceSettings.ThreadAutoFollow && post.RootId != "" { - nErr := a.Srv().Store.Thread().CreateMembershipIfNeeded(userId, post.RootId) + nErr := a.Srv().Store.Thread().CreateMembershipIfNeeded(userId, post.RootId, true) if nErr != nil { mac <- model.NewAppError("SendNotifications", "app.channel.autofollow.app_error", nil, nErr.Error(), http.StatusInternalServerError) return diff --git a/app/opentracing/opentracing_layer.go b/app/opentracing/opentracing_layer.go index 16a2ec50cd..b74433b718 100644 --- a/app/opentracing/opentracing_layer.go +++ b/app/opentracing/opentracing_layer.go @@ -8441,6 +8441,28 @@ func (a *OpenTracingAppLayer) GetThreadMembershipsForUser(userId string) ([]*mod return resultVar0, resultVar1 } +func (a *OpenTracingAppLayer) GetThreadsForUser(userId string, options model.GetUserThreadsOpts) (*model.Threads, *model.AppError) { + origCtx := a.ctx + span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.GetThreadsForUser") + + a.ctx = newCtx + a.app.Srv().Store.SetContext(newCtx) + defer func() { + a.app.Srv().Store.SetContext(origCtx) + a.ctx = origCtx + }() + + defer span.Finish() + resultVar0, resultVar1 := a.app.GetThreadsForUser(userId, options) + + if resultVar1 != nil { + span.LogFields(spanlog.Error(resultVar1)) + ext.Error.Set(span, true) + } + + return resultVar0, resultVar1 +} + func (a *OpenTracingAppLayer) GetTotalUsersStats(viewRestrictions *model.ViewUsersRestrictions) (*model.UsersStats, *model.AppError) { origCtx := a.ctx span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.GetTotalUsersStats") @@ -15138,6 +15160,72 @@ func (a *OpenTracingAppLayer) UpdateTeamScheme(team *model.Team) (*model.Team, * return resultVar0, resultVar1 } +func (a *OpenTracingAppLayer) UpdateThreadFollowForUser(userId string, threadId string, state bool) *model.AppError { + origCtx := a.ctx + span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.UpdateThreadFollowForUser") + + 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.UpdateThreadFollowForUser(userId, threadId, state) + + if resultVar0 != nil { + span.LogFields(spanlog.Error(resultVar0)) + ext.Error.Set(span, true) + } + + return resultVar0 +} + +func (a *OpenTracingAppLayer) UpdateThreadReadForUser(userId string, threadId string, timestamp int64) *model.AppError { + origCtx := a.ctx + span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.UpdateThreadReadForUser") + + 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.UpdateThreadReadForUser(userId, threadId, timestamp) + + if resultVar0 != nil { + span.LogFields(spanlog.Error(resultVar0)) + ext.Error.Set(span, true) + } + + return resultVar0 +} + +func (a *OpenTracingAppLayer) UpdateThreadsReadForUser(userId string, timestamp int64) *model.AppError { + origCtx := a.ctx + span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.UpdateThreadsReadForUser") + + 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.UpdateThreadsReadForUser(userId, timestamp) + + if resultVar0 != nil { + span.LogFields(spanlog.Error(resultVar0)) + ext.Error.Set(span, true) + } + + return resultVar0 +} + func (a *OpenTracingAppLayer) UpdateUser(user *model.User, sendNotifications bool) (*model.User, *model.AppError) { origCtx := a.ctx span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.UpdateUser") diff --git a/app/post.go b/app/post.go index f6810c9754..5175db6a98 100644 --- a/app/post.go +++ b/app/post.go @@ -458,7 +458,7 @@ func (a *App) handlePostEvents(post *model.Post, user *model.User, channel *mode } if *a.Config().ServiceSettings.ThreadAutoFollow && post.RootId != "" { - if err := a.Srv().Store.Thread().CreateMembershipIfNeeded(post.UserId, post.RootId); err != nil { + if err := a.Srv().Store.Thread().CreateMembershipIfNeeded(post.UserId, post.RootId, true); err != nil { return err } } diff --git a/app/user.go b/app/user.go index c80485df51..ef997c4f28 100644 --- a/app/user.go +++ b/app/user.go @@ -2368,3 +2368,39 @@ func (a *App) ConvertBotToUser(bot *model.Bot, userPatch *model.UserPatch, sysad return user, nil } + +func (a *App) GetThreadsForUser(userId string, options model.GetUserThreadsOpts) (*model.Threads, *model.AppError) { + threads, err := a.Srv().Store.Thread().GetThreadsForUser(userId, options) + if err != nil { + return nil, model.NewAppError("GetThreadsForUser", "app.user.get_threads_for_user.app_error", nil, err.Error(), http.StatusInternalServerError) + } + for _, thread := range threads.Threads { + a.sanitizeProfiles(thread.Participants, false) + thread.Post.SanitizeProps() + } + return threads, nil +} + +func (a *App) UpdateThreadsReadForUser(userId string, timestamp int64) *model.AppError { + err := a.Srv().Store.Thread().MarkAllAsRead(userId, timestamp) + if err != nil { + return model.NewAppError("UpdateThreadsReadForUser", "app.user.update_threads_read_for_user.app_error", nil, err.Error(), http.StatusInternalServerError) + } + return nil +} + +func (a *App) UpdateThreadFollowForUser(userId, threadId string, state bool) *model.AppError { + err := a.Srv().Store.Thread().CreateMembershipIfNeeded(userId, threadId, state) + if err != nil { + return model.NewAppError("UpdateThreadFollowForUser", "app.user.update_thread_follow_for_user.app_error", nil, err.Error(), http.StatusInternalServerError) + } + return nil +} + +func (a *App) UpdateThreadReadForUser(userId, threadId string, timestamp int64) *model.AppError { + err := a.Srv().Store.Thread().MarkAsRead(userId, threadId, timestamp) + if err != nil { + return model.NewAppError("UpdateThreadReadForUser", "app.user.update_thread_read_for_user.app_error", nil, err.Error(), http.StatusInternalServerError) + } + return nil +} diff --git a/i18n/en.json b/i18n/en.json index 128c9107d7..56b2ed046a 100644 --- a/i18n/en.json +++ b/i18n/en.json @@ -5434,6 +5434,10 @@ "id": "app.user.get_recently_active_users.app_error", "translation": "We encountered an error while finding the recently active users." }, + { + "id": "app.user.get_threads_for_user.app_error", + "translation": "Unable to get user threads" + }, { "id": "app.user.get_total_users_count.app_error", "translation": "We could not count the users." @@ -5510,6 +5514,18 @@ "id": "app.user.update_password.app_error", "translation": "Unable to update the user password." }, + { + "id": "app.user.update_thread_follow_for_user.app_error", + "translation": "Unable to update following state for thread" + }, + { + "id": "app.user.update_thread_read_for_user.app_error", + "translation": "Unable to update read state for thread" + }, + { + "id": "app.user.update_threads_read_for_user.app_error", + "translation": "Unable to update all user threads as read" + }, { "id": "app.user.update_update.app_error", "translation": "Unable to update the date of the last update of the user." @@ -5824,15 +5840,15 @@ }, { "id": "ent.cloud.authentication_failed", - "translation": "Unable to authenticate to CWS" + "translation": "" }, { "id": "ent.cloud.json_encode.error", - "translation": "Internal error marshaling request to CWS" + "translation": "" }, { "id": "ent.cloud.request_error", - "translation": "Error processing request to CWS" + "translation": "" }, { "id": "ent.cluster.404.app_error", @@ -5944,7 +5960,7 @@ }, { "id": "ent.data_retention.posts_permanent_delete_batch.internal_error", - "translation": "We encountered an error permanently deleting the batch of posts." + "translation": "" }, { "id": "ent.data_retention.reactions_batch.internal_error", @@ -6020,7 +6036,7 @@ }, { "id": "ent.elasticsearch.index_channels_batch.error", - "translation": "Unable to get the channels batch for indexing." + "translation": "" }, { "id": "ent.elasticsearch.index_post.error", @@ -6052,7 +6068,7 @@ }, { "id": "ent.elasticsearch.post.get_posts_batch_for_indexing.error", - "translation": "Unable to get the posts batch for indexing." + "translation": "" }, { "id": "ent.elasticsearch.purge_indexes.delete_failed", @@ -6412,11 +6428,11 @@ }, { "id": "ent.saml.do_login.invalid_signature.app_error", - "translation": "We received an invalid signature in the response from the Identity Provider. Please contact your System Administrator." + "translation": "" }, { "id": "ent.saml.do_login.invalid_time.app_error", - "translation": "We received an invalid time in the response from the Identity Provider. Please contact your System Administrator." + "translation": "" }, { "id": "ent.saml.do_login.parse.app_error", diff --git a/model/client4.go b/model/client4.go index be0ea2b8a6..8ebf0da162 100644 --- a/model/client4.go +++ b/model/client4.go @@ -189,6 +189,14 @@ func (c *Client4) GetUserRoute(userId string) string { return fmt.Sprintf(c.GetUsersRoute()+"/%v", userId) } +func (c *Client4) GetUserThreadsRoute(userId string) string { + return fmt.Sprintf(c.GetUsersRoute()+"/%v/threads", userId) +} + +func (c *Client4) GetUserThreadRoute(userId, threadId string) string { + return fmt.Sprintf(c.GetUserThreadsRoute(userId)+"/%v", threadId) +} + func (c *Client4) GetUserCategoryRoute(userID, teamID string) string { return c.GetUserRoute(userID) + c.GetTeamRoute(teamID) + "/channels/categories" } @@ -5744,3 +5752,74 @@ func (c *Client4) UpdateCloudCustomerAddress(address *Address) (*CloudCustomer, return customer, BuildResponse(r) } + +func (c *Client4) GetUserThreads(userId string, options GetUserThreadsOpts) (*Threads, *Response) { + v := url.Values{} + if options.Since != 0 { + v.Set("since", fmt.Sprintf("%d", options.Since)) + } + if options.Page != 0 { + v.Set("page", fmt.Sprintf("%d", options.Page)) + } + if options.PageSize != 0 { + v.Set("pageSize", fmt.Sprintf("%d", options.PageSize)) + } + if options.Extended { + v.Set("extended", "true") + } + if options.Deleted { + v.Set("deleted", "true") + } + + url := c.GetUserThreadsRoute(userId) + if len(v) > 0 { + url += "?" + v.Encode() + } + + r, appErr := c.DoApiGet(url, "") + if appErr != nil { + return nil, BuildErrorResponse(r, appErr) + } + defer closeBody(r) + + var threads Threads + json.NewDecoder(r.Body).Decode(&threads) + + return &threads, BuildResponse(r) +} + +func (c *Client4) UpdateThreadsReadForUser(userId string, timestamp int64) *Response { + r, appErr := c.DoApiPut(fmt.Sprintf("%s/read/%d", c.GetUserThreadsRoute(userId), timestamp), "") + if appErr != nil { + return BuildErrorResponse(r, appErr) + } + defer closeBody(r) + + return BuildResponse(r) +} + +func (c *Client4) UpdateThreadReadForUser(userId, threadId string, timestamp int64) *Response { + r, appErr := c.DoApiPut(fmt.Sprintf("%s/read/%d", c.GetUserThreadRoute(userId, threadId), timestamp), "") + if appErr != nil { + return BuildErrorResponse(r, appErr) + } + defer closeBody(r) + + return BuildResponse(r) +} + +func (c *Client4) UpdateThreadFollowForUser(userId, threadId string, state bool) *Response { + var appErr *AppError + var r *http.Response + if state { + r, appErr = c.DoApiPut(c.GetUserThreadRoute(userId, threadId)+"/following", "") + } else { + r, appErr = c.DoApiDelete(c.GetUserThreadRoute(userId, threadId) + "/following") + } + if appErr != nil { + return BuildErrorResponse(r, appErr) + } + defer closeBody(r) + + return BuildResponse(r) +} diff --git a/model/thread.go b/model/thread.go index 969a599e1e..80e655b8ef 100644 --- a/model/thread.go +++ b/model/thread.go @@ -15,6 +15,42 @@ type Thread struct { Participants StringArray `json:"participants"` } +type ThreadResponse struct { + PostId string `json:"id"` + ReplyCount int64 `json:"reply_count"` + LastReplyAt int64 `json:"last_reply_at"` + LastViewedAt int64 `json:"last_viewed_at"` + Participants []*User `json:"participants"` + Post *Post `json:"post"` +} + +type Threads struct { + Total int64 `json:"total"` + Threads []*ThreadResponse `json:"threads"` +} + +type GetUserThreadsOpts struct { + // Page specifies which part of the results to return, by PageSize. Default = 0 + Page uint64 + + // PageSize specifies the size of the returned chunk of results. Default = 30 + PageSize uint64 + + // Extended will enrich the response with participant details. Default = false + Extended bool + + // Deleted will specify that even deleted threads should be returned (For mobile sync). Default = false + Deleted bool + + // Since filters the threads based on their LastUpdateAt timestamp. + Since uint64 +} + +func (o *Threads) ToJson() string { + b, _ := json.Marshal(o) + return string(b) +} + func (o *Thread) ToJson() string { b, _ := json.Marshal(o) return string(b) diff --git a/store/opentracinglayer/opentracinglayer.go b/store/opentracinglayer/opentracinglayer.go index 9b3b6d9d6d..e4ae40d289 100644 --- a/store/opentracinglayer/opentracinglayer.go +++ b/store/opentracinglayer/opentracinglayer.go @@ -7648,7 +7648,7 @@ func (s *OpenTracingLayerThreadStore) CollectThreadsWithNewerReplies(userId stri return result, err } -func (s *OpenTracingLayerThreadStore) CreateMembershipIfNeeded(userId string, postId string) error { +func (s *OpenTracingLayerThreadStore) CreateMembershipIfNeeded(userId string, postId string, following bool) error { origCtx := s.Root.Store.Context() span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ThreadStore.CreateMembershipIfNeeded") s.Root.Store.SetContext(newCtx) @@ -7657,7 +7657,7 @@ func (s *OpenTracingLayerThreadStore) CreateMembershipIfNeeded(userId string, po }() defer span.Finish() - err := s.ThreadStore.CreateMembershipIfNeeded(userId, postId) + err := s.ThreadStore.CreateMembershipIfNeeded(userId, postId, following) if err != nil { span.LogFields(spanlog.Error(err)) ext.Error.Set(span, true) @@ -7756,6 +7756,60 @@ func (s *OpenTracingLayerThreadStore) GetMembershipsForUser(userId string) ([]*m return result, err } +func (s *OpenTracingLayerThreadStore) GetThreadsForUser(userId string, opts model.GetUserThreadsOpts) (*model.Threads, error) { + origCtx := s.Root.Store.Context() + span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ThreadStore.GetThreadsForUser") + s.Root.Store.SetContext(newCtx) + defer func() { + s.Root.Store.SetContext(origCtx) + }() + + defer span.Finish() + result, err := s.ThreadStore.GetThreadsForUser(userId, opts) + if err != nil { + span.LogFields(spanlog.Error(err)) + ext.Error.Set(span, true) + } + + return result, err +} + +func (s *OpenTracingLayerThreadStore) MarkAllAsRead(userId string, timestamp int64) error { + origCtx := s.Root.Store.Context() + span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ThreadStore.MarkAllAsRead") + s.Root.Store.SetContext(newCtx) + defer func() { + s.Root.Store.SetContext(origCtx) + }() + + defer span.Finish() + err := s.ThreadStore.MarkAllAsRead(userId, timestamp) + if err != nil { + span.LogFields(spanlog.Error(err)) + ext.Error.Set(span, true) + } + + return err +} + +func (s *OpenTracingLayerThreadStore) MarkAsRead(userId string, threadId string, timestamp int64) error { + origCtx := s.Root.Store.Context() + span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ThreadStore.MarkAsRead") + s.Root.Store.SetContext(newCtx) + defer func() { + s.Root.Store.SetContext(origCtx) + }() + + defer span.Finish() + err := s.ThreadStore.MarkAsRead(userId, threadId, timestamp) + if err != nil { + span.LogFields(spanlog.Error(err)) + ext.Error.Set(span, true) + } + + return err +} + func (s *OpenTracingLayerThreadStore) Save(thread *model.Thread) (*model.Thread, error) { origCtx := s.Root.Store.Context() span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ThreadStore.Save") diff --git a/store/retrylayer/retrylayer.go b/store/retrylayer/retrylayer.go index 4b98e79015..4b9b56b364 100644 --- a/store/retrylayer/retrylayer.go +++ b/store/retrylayer/retrylayer.go @@ -7682,11 +7682,11 @@ func (s *RetryLayerThreadStore) CollectThreadsWithNewerReplies(userId string, ch } -func (s *RetryLayerThreadStore) CreateMembershipIfNeeded(userId string, postId string) error { +func (s *RetryLayerThreadStore) CreateMembershipIfNeeded(userId string, postId string, following bool) error { tries := 0 for { - err := s.ThreadStore.CreateMembershipIfNeeded(userId, postId) + err := s.ThreadStore.CreateMembershipIfNeeded(userId, postId, following) if err == nil { return nil } @@ -7802,6 +7802,66 @@ func (s *RetryLayerThreadStore) GetMembershipsForUser(userId string) ([]*model.T } +func (s *RetryLayerThreadStore) GetThreadsForUser(userId string, opts model.GetUserThreadsOpts) (*model.Threads, error) { + + tries := 0 + for { + result, err := s.ThreadStore.GetThreadsForUser(userId, opts) + 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 + } + } + +} + +func (s *RetryLayerThreadStore) MarkAllAsRead(userId string, timestamp int64) error { + + tries := 0 + for { + err := s.ThreadStore.MarkAllAsRead(userId, timestamp) + 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 + } + } + +} + +func (s *RetryLayerThreadStore) MarkAsRead(userId string, threadId string, timestamp int64) error { + + tries := 0 + for { + err := s.ThreadStore.MarkAsRead(userId, threadId, timestamp) + 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 + } + } + +} + func (s *RetryLayerThreadStore) Save(thread *model.Thread) (*model.Thread, error) { tries := 0 diff --git a/store/sqlstore/post_store.go b/store/sqlstore/post_store.go index 7df9f8f4be..89d60c50ff 100644 --- a/store/sqlstore/post_store.go +++ b/store/sqlstore/post_store.go @@ -515,20 +515,17 @@ func (s *SqlPostStore) Delete(postId string, time int64, deleteByID string) erro return errors.Wrap(err, "failed to update Posts") } - return s.cleanupThreads(post.Id, post.RootId, post.UserId) + return s.cleanupThreads(post.Id, post.RootId, post.UserId, false) } func (s *SqlPostStore) permanentDelete(postId string) error { var post model.Post err := s.GetReplica().SelectOne(&post, "SELECT * FROM Posts WHERE Id = :Id AND DeleteAt = 0", map[string]interface{}{"Id": postId}) if err != nil && err != sql.ErrNoRows { - if err != sql.ErrNoRows { - return errors.Wrapf(err, "failed to get Post with id=%s", postId) - } - - if err = s.cleanupThreads(post.Id, post.RootId, post.UserId); err != nil { - return errors.Wrapf(err, "failed to cleanup threads for Post with id=%s", postId) - } + return errors.Wrapf(err, "failed to get Post with id=%s", postId) + } + if err = s.cleanupThreads(post.Id, post.RootId, post.UserId, true); err != nil { + return errors.Wrapf(err, "failed to cleanup threads for Post with id=%s", postId) } if _, err = s.GetMaster().Exec("DELETE FROM Posts WHERE Id = :Id OR RootId = :RootId", map[string]interface{}{"Id": postId, "RootId": postId}); err != nil { @@ -552,7 +549,7 @@ func (s *SqlPostStore) permanentDeleteAllCommentByUser(userId string) error { } for _, ids := range results { - if err = s.cleanupThreads(ids.Id, ids.RootId, userId); err != nil { + if err = s.cleanupThreads(ids.Id, ids.RootId, userId, true); err != nil { return err } } @@ -608,7 +605,7 @@ func (s *SqlPostStore) PermanentDeleteByChannel(channelId string) error { } for _, ids := range results { - if err = s.cleanupThreads(ids.Id, ids.RootId, ids.UserId); err != nil { + if err = s.cleanupThreads(ids.Id, ids.RootId, ids.UserId, true); err != nil { return err } } @@ -1921,7 +1918,16 @@ func (s *SqlPostStore) GetOldestEntityCreationTime() (int64, error) { return oldest, nil } -func (s *SqlPostStore) cleanupThreads(postId, rootId, userId string) error { +func (s *SqlPostStore) cleanupThreads(postId, rootId, userId string, permanent bool) error { + if permanent { + if _, err := s.GetMaster().Exec("DELETE FROM Threads WHERE PostId = :Id", map[string]interface{}{"Id": postId}); err != nil { + return errors.Wrap(err, "failed to delete Threads") + } + if _, err := s.GetMaster().Exec("DELETE FROM ThreadMemberships WHERE PostId = :Id", map[string]interface{}{"Id": postId}); err != nil { + return errors.Wrap(err, "failed to delete ThreadMemberships") + } + return nil + } if len(rootId) > 0 { thread, err := s.Thread().Get(rootId) if err != nil { @@ -1937,12 +1943,6 @@ func (s *SqlPostStore) cleanupThreads(postId, rootId, userId string) error { } } } - if _, err := s.GetMaster().Exec("DELETE FROM Threads WHERE PostId = :Id", map[string]interface{}{"Id": postId}); err != nil { - return errors.Wrap(err, "failed to delete Threads") - } - if _, err := s.GetMaster().Exec("DELETE FROM ThreadMemberships WHERE PostId = :Id", map[string]interface{}{"Id": postId}); err != nil { - return errors.Wrap(err, "failed to delete ThreadMemberships") - } return nil } diff --git a/store/sqlstore/thread_store.go b/store/sqlstore/thread_store.go index 58a5e319d7..c6d63288de 100644 --- a/store/sqlstore/thread_store.go +++ b/store/sqlstore/thread_store.go @@ -108,6 +108,117 @@ func (s *SqlThreadStore) Get(id string) (*model.Thread, error) { return &thread, nil } +func (s *SqlThreadStore) GetThreadsForUser(userId string, opts model.GetUserThreadsOpts) (*model.Threads, error) { + type JoinedThread struct { + PostId string + ReplyCount int64 + LastReplyAt int64 + LastViewedAt int64 + Participants model.StringArray + model.Post + } + var threads []*JoinedThread + + fetchConditions := sq.And{ + sq.Eq{"Posts.UserId": userId}, + sq.Eq{"ThreadMemberships.Following": true}, + } + if !opts.Deleted { + fetchConditions = sq.And{fetchConditions, sq.Eq{"Posts.DeleteAt": 0}} + } + if opts.Since > 0 { + fetchConditions = sq.And{fetchConditions, sq.GtOrEq{"Threads.LastReplyAt": opts.Since}} + } + pageSize := uint64(30) + if opts.PageSize == 0 { + pageSize = opts.PageSize + } + query, args, _ := s.getQueryBuilder(). + Select("Threads.*, Posts.*, ThreadMemberships.LastViewed as LastViewedAt"). + From("Threads"). + LeftJoin("Posts ON Posts.Id = Threads.PostId"). + LeftJoin("ThreadMemberships ON ThreadMemberships.PostId = Threads.PostId"). + OrderBy("Threads.LastReplyAt DESC"). + Offset(pageSize * opts.Page). + Limit(pageSize). + Where(fetchConditions).ToSql() + _, err := s.GetReplica().Select(&threads, query, args...) + if err != nil { + return nil, errors.Wrapf(err, "failed to get threads for user id=%s", userId) + } + + var userIds []string + userIdMap := map[string]bool{} + for _, thread := range threads { + for _, participantId := range thread.Participants { + if _, ok := userIdMap[participantId]; !ok { + userIdMap[participantId] = true + userIds = append(userIds, participantId) + } + } + } + var users []*model.User + if opts.Extended { + query, args, _ = s.getQueryBuilder().Select("*").From("Users").Where(sq.Eq{"Id": userIds}).ToSql() + _, err = s.GetReplica().Select(&users, query, args...) + if err != nil { + return nil, errors.Wrapf(err, "failed to get threads for user id=%s", userId) + } + } else { + for _, userId := range userIds { + users = append(users, &model.User{Id: userId}) + } + } + + result := &model.Threads{ + Total: 0, + Threads: nil, + } + + for _, thread := range threads { + var participants []*model.User + for _, participantId := range thread.Participants { + var participant *model.User + for _, u := range users { + if u.Id == participantId { + participant = u + break + } + } + if participant == nil { + return nil, errors.New("cannot find thread participant with id=" + participantId) + } + participants = append(participants, participant) + } + result.Threads = append(result.Threads, &model.ThreadResponse{ + PostId: thread.PostId, + ReplyCount: thread.ReplyCount, + LastReplyAt: thread.LastReplyAt, + LastViewedAt: thread.LastViewedAt, + Participants: participants, + Post: &thread.Post, + }) + } + + return result, nil +} + +func (s *SqlThreadStore) MarkAllAsRead(userId string, timestamp int64) error { + query, args, _ := s.getQueryBuilder().Update("ThreadMemberships").Where(sq.Eq{"UserId": userId}).Set("LastViewed", timestamp).ToSql() + if _, err := s.GetMaster().Exec(query, args...); err != nil { + return errors.Wrapf(err, "failed to update thread read state for user id=%s", userId) + } + return nil +} + +func (s *SqlThreadStore) MarkAsRead(userId, threadId string, timestamp int64) error { + query, args, _ := s.getQueryBuilder().Update("ThreadMemberships").Where(sq.Eq{"UserId": userId}, sq.Eq{"PostId": threadId}).Set("LastViewed", timestamp).ToSql() + if _, err := s.GetMaster().Exec(query, args...); err != nil { + return errors.Wrapf(err, "failed to update thread read state for user id=%s thread_id=%v", userId, threadId) + } + return nil +} + func (s *SqlThreadStore) Delete(threadId string) error { query, args, _ := s.getQueryBuilder().Delete("Threads").Where(sq.Eq{"PostId": threadId}).ToSql() if _, err := s.GetMaster().Exec(query, args...); err != nil { @@ -162,12 +273,12 @@ func (s *SqlThreadStore) DeleteMembershipForUser(userId string, postId string) e return nil } -func (s *SqlThreadStore) CreateMembershipIfNeeded(userId, postId string) error { +func (s *SqlThreadStore) CreateMembershipIfNeeded(userId, postId string, following bool) error { membership, err := s.GetMembershipForUser(userId, postId) now := utils.MillisFromTime(time.Now()) if err == nil { - if !membership.Following { - membership.Following = true + if !membership.Following || membership.Following != following { + membership.Following = following membership.LastUpdated = now _, err = s.UpdateMembership(membership) } @@ -182,7 +293,7 @@ func (s *SqlThreadStore) CreateMembershipIfNeeded(userId, postId string) error { _, err = s.SaveMembership(&model.ThreadMembership{ PostId: postId, UserId: userId, - Following: true, + Following: following, LastViewed: 0, LastUpdated: now, }) diff --git a/store/store.go b/store/store.go index d7c2a293eb..73b75e2f5a 100644 --- a/store/store.go +++ b/store/store.go @@ -251,14 +251,18 @@ type ThreadStore interface { Save(thread *model.Thread) (*model.Thread, error) Update(thread *model.Thread) (*model.Thread, error) Get(id string) (*model.Thread, error) + GetThreadsForUser(userId string, opts model.GetUserThreadsOpts) (*model.Threads, error) Delete(postId string) error + MarkAllAsRead(userId string, timestamp int64) error + MarkAsRead(userId, threadId string, timestamp int64) error + SaveMembership(membership *model.ThreadMembership) (*model.ThreadMembership, error) UpdateMembership(membership *model.ThreadMembership) (*model.ThreadMembership, error) GetMembershipsForUser(userId string) ([]*model.ThreadMembership, error) GetMembershipForUser(userId, postId string) (*model.ThreadMembership, error) DeleteMembershipForUser(userId, postId string) error - CreateMembershipIfNeeded(userId, postId string) error + CreateMembershipIfNeeded(userId, postId string, following bool) error CollectThreadsWithNewerReplies(userId string, channelIds []string, timestamp int64) ([]string, error) UpdateUnreadsByChannel(userId string, changedThreads []string, timestamp int64) error } diff --git a/store/storetest/mocks/ThreadStore.go b/store/storetest/mocks/ThreadStore.go index e8d762e152..c56a709722 100644 --- a/store/storetest/mocks/ThreadStore.go +++ b/store/storetest/mocks/ThreadStore.go @@ -37,13 +37,13 @@ func (_m *ThreadStore) CollectThreadsWithNewerReplies(userId string, channelIds return r0, r1 } -// CreateMembershipIfNeeded provides a mock function with given fields: userId, postId -func (_m *ThreadStore) CreateMembershipIfNeeded(userId string, postId string) error { - ret := _m.Called(userId, postId) +// CreateMembershipIfNeeded provides a mock function with given fields: userId, postId, following +func (_m *ThreadStore) CreateMembershipIfNeeded(userId string, postId string, following bool) error { + ret := _m.Called(userId, postId, following) var r0 error - if rf, ok := ret.Get(0).(func(string, string) error); ok { - r0 = rf(userId, postId) + if rf, ok := ret.Get(0).(func(string, string, bool) error); ok { + r0 = rf(userId, postId, following) } else { r0 = ret.Error(0) } @@ -148,6 +148,57 @@ func (_m *ThreadStore) GetMembershipsForUser(userId string) ([]*model.ThreadMemb return r0, r1 } +// GetThreadsForUser provides a mock function with given fields: userId, opts +func (_m *ThreadStore) GetThreadsForUser(userId string, opts model.GetUserThreadsOpts) (*model.Threads, error) { + ret := _m.Called(userId, opts) + + var r0 *model.Threads + if rf, ok := ret.Get(0).(func(string, model.GetUserThreadsOpts) *model.Threads); ok { + r0 = rf(userId, opts) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).(*model.Threads) + } + } + + var r1 error + if rf, ok := ret.Get(1).(func(string, model.GetUserThreadsOpts) error); ok { + r1 = rf(userId, opts) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + +// MarkAllAsRead provides a mock function with given fields: userId, timestamp +func (_m *ThreadStore) MarkAllAsRead(userId string, timestamp int64) error { + ret := _m.Called(userId, timestamp) + + var r0 error + if rf, ok := ret.Get(0).(func(string, int64) error); ok { + r0 = rf(userId, timestamp) + } else { + r0 = ret.Error(0) + } + + return r0 +} + +// MarkAsRead provides a mock function with given fields: userId, threadId, timestamp +func (_m *ThreadStore) MarkAsRead(userId string, threadId string, timestamp int64) error { + ret := _m.Called(userId, threadId, timestamp) + + var r0 error + if rf, ok := ret.Get(0).(func(string, string, int64) error); ok { + r0 = rf(userId, threadId, timestamp) + } else { + r0 = ret.Error(0) + } + + return r0 +} + // Save provides a mock function with given fields: thread func (_m *ThreadStore) Save(thread *model.Thread) (*model.Thread, error) { ret := _m.Called(thread) diff --git a/store/storetest/thread_store.go b/store/storetest/thread_store.go index deed3f3db4..fa9e6ed34c 100644 --- a/store/storetest/thread_store.go +++ b/store/storetest/thread_store.go @@ -223,7 +223,7 @@ func testThreadStorePopulation(t *testing.T, ss store.Store) { require.EqualValues(t, thread1.ReplyCount, 1) require.Len(t, thread1.Participants, 1) - err = ss.Post().Delete(rootPost.Id, 123, model.NewId()) + err = ss.Post().PermanentDeleteByUser(rootPost.UserId) require.Nil(t, err) thread2, _ := ss.Thread().Get(rootPost.Id) @@ -233,7 +233,7 @@ func testThreadStorePopulation(t *testing.T, ss store.Store) { t.Run("Thread last updated is changed when channel is updated after UpdateLastViewedAtPost", func(t *testing.T) { newPosts := makeSomePosts() - require.Nil(t, ss.Thread().CreateMembershipIfNeeded(newPosts[0].UserId, newPosts[0].Id)) + require.Nil(t, ss.Thread().CreateMembershipIfNeeded(newPosts[0].UserId, newPosts[0].Id, true)) m, err1 := ss.Thread().GetMembershipForUser(newPosts[0].UserId, newPosts[0].Id) require.Nil(t, err1) m.LastUpdated -= 1000 @@ -253,7 +253,7 @@ func testThreadStorePopulation(t *testing.T, ss store.Store) { t.Run("Thread last updated is changed when channel is updated after IncrementMentionCount", func(t *testing.T) { newPosts := makeSomePosts() - require.Nil(t, ss.Thread().CreateMembershipIfNeeded(newPosts[0].UserId, newPosts[0].Id)) + require.Nil(t, ss.Thread().CreateMembershipIfNeeded(newPosts[0].UserId, newPosts[0].Id, true)) m, err1 := ss.Thread().GetMembershipForUser(newPosts[0].UserId, newPosts[0].Id) require.Nil(t, err1) m.LastUpdated -= 1000 @@ -273,7 +273,7 @@ func testThreadStorePopulation(t *testing.T, ss store.Store) { t.Run("Thread last updated is changed when channel is updated after UpdateLastViewedAt", func(t *testing.T) { newPosts := makeSomePosts() - require.Nil(t, ss.Thread().CreateMembershipIfNeeded(newPosts[0].UserId, newPosts[0].Id)) + require.Nil(t, ss.Thread().CreateMembershipIfNeeded(newPosts[0].UserId, newPosts[0].Id, true)) m, err1 := ss.Thread().GetMembershipForUser(newPosts[0].UserId, newPosts[0].Id) require.Nil(t, err1) m.LastUpdated -= 1000 @@ -293,7 +293,7 @@ func testThreadStorePopulation(t *testing.T, ss store.Store) { t.Run("Thread last updated is changed when channel is updated after UpdateLastViewedAtPost for mark unread", func(t *testing.T) { newPosts := makeSomePosts() - require.Nil(t, ss.Thread().CreateMembershipIfNeeded(newPosts[0].UserId, newPosts[0].Id)) + require.Nil(t, ss.Thread().CreateMembershipIfNeeded(newPosts[0].UserId, newPosts[0].Id, true)) m, err1 := ss.Thread().GetMembershipForUser(newPosts[0].UserId, newPosts[0].Id) require.Nil(t, err1) m.LastUpdated += 1000 diff --git a/store/timerlayer/timerlayer.go b/store/timerlayer/timerlayer.go index 89d8de4a87..17a206689e 100644 --- a/store/timerlayer/timerlayer.go +++ b/store/timerlayer/timerlayer.go @@ -6904,10 +6904,10 @@ func (s *TimerLayerThreadStore) CollectThreadsWithNewerReplies(userId string, ch return result, err } -func (s *TimerLayerThreadStore) CreateMembershipIfNeeded(userId string, postId string) error { +func (s *TimerLayerThreadStore) CreateMembershipIfNeeded(userId string, postId string, following bool) error { start := timemodule.Now() - err := s.ThreadStore.CreateMembershipIfNeeded(userId, postId) + err := s.ThreadStore.CreateMembershipIfNeeded(userId, postId, following) elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second) if s.Root.Metrics != nil { @@ -7000,6 +7000,54 @@ func (s *TimerLayerThreadStore) GetMembershipsForUser(userId string) ([]*model.T return result, err } +func (s *TimerLayerThreadStore) GetThreadsForUser(userId string, opts model.GetUserThreadsOpts) (*model.Threads, error) { + start := timemodule.Now() + + result, err := s.ThreadStore.GetThreadsForUser(userId, opts) + + elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second) + if s.Root.Metrics != nil { + success := "false" + if err == nil { + success = "true" + } + s.Root.Metrics.ObserveStoreMethodDuration("ThreadStore.GetThreadsForUser", success, elapsed) + } + return result, err +} + +func (s *TimerLayerThreadStore) MarkAllAsRead(userId string, timestamp int64) error { + start := timemodule.Now() + + err := s.ThreadStore.MarkAllAsRead(userId, timestamp) + + elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second) + if s.Root.Metrics != nil { + success := "false" + if err == nil { + success = "true" + } + s.Root.Metrics.ObserveStoreMethodDuration("ThreadStore.MarkAllAsRead", success, elapsed) + } + return err +} + +func (s *TimerLayerThreadStore) MarkAsRead(userId string, threadId string, timestamp int64) error { + start := timemodule.Now() + + err := s.ThreadStore.MarkAsRead(userId, threadId, timestamp) + + elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second) + if s.Root.Metrics != nil { + success := "false" + if err == nil { + success = "true" + } + s.Root.Metrics.ObserveStoreMethodDuration("ThreadStore.MarkAsRead", success, elapsed) + } + return err +} + func (s *TimerLayerThreadStore) Save(thread *model.Thread) (*model.Thread, error) { start := timemodule.Now() diff --git a/web/context.go b/web/context.go index 34d5d8a23b..28108e8b8c 100644 --- a/web/context.go +++ b/web/context.go @@ -329,6 +329,28 @@ func (c *Context) RequireTokenId() *Context { return c } +func (c *Context) RequireThreadId() *Context { + if c.Err != nil { + return c + } + + if !model.IsValidId(c.Params.ThreadId) { + c.SetInvalidUrlParam("thread_id") + } + return c +} + +func (c *Context) RequireTimestamp() *Context { + if c.Err != nil { + return c + } + + if c.Params.Timestamp == 0 { + c.SetInvalidUrlParam("timestamp") + } + return c +} + func (c *Context) RequireChannelId() *Context { if c.Err != nil { return c diff --git a/web/params.go b/web/params.go index 53043495bd..a3baf465ff 100644 --- a/web/params.go +++ b/web/params.go @@ -27,6 +27,8 @@ type Params struct { TeamId string InviteId string TokenId string + ThreadId string + Timestamp int64 ChannelId string PostId string FileId string @@ -111,6 +113,10 @@ func ParamsFromRequest(r *http.Request) *Params { params.TokenId = val } + if val, ok := props["thread_id"]; ok { + params.ThreadId = val + } + if val, ok := props["channel_id"]; ok { params.ChannelId = val } else { @@ -231,6 +237,12 @@ func ParamsFromRequest(r *http.Request) *Params { params.Page = val } + if val, err := strconv.ParseInt(props["timestamp"], 10, 64); err != nil || val < 0 { + params.Timestamp = 0 + } else { + params.Timestamp = val + } + if val, err := strconv.ParseBool(query.Get("permanent")); err == nil { params.Permanent = val } diff --git a/web/web_test.go b/web/web_test.go index 0a1cdeec8e..9799eff6a5 100644 --- a/web/web_test.go +++ b/web/web_test.go @@ -73,7 +73,6 @@ func setupTestHelper(t testing.TB, store store.Store, includeCacheLayer bool) *T *newConfig.AnnouncementSettings.AdminNoticesEnabled = false *newConfig.AnnouncementSettings.UserNoticesEnabled = false memoryStore.Set(newConfig) - var options []app.Option options = append(options, app.ConfigStore(memoryStore)) options = append(options, app.StoreOverride(mainHelper.Store))