Coverage for app/backend/src/couchers/servicers/discussions.py: 87%

1import logging 

2 

3import grpc 

4from google.protobuf import empty_pb2 

5from sqlalchemy import select 

6from sqlalchemy.orm import Session, aliased 

7 

8from couchers.context import CouchersContext, make_notification_user_context 

9from couchers.db import can_moderate_node, session_scope 

10from couchers.event_log import log_event 

11from couchers.helpers.completed_profile import has_completed_profile 

12from couchers.jobs.enqueue import queue_job 

13from couchers.models import Cluster, ClusterSubscription, Discussion, ModerationObjectType, Thread, User 

14from couchers.models.discussions import ContentChangeType, DiscussionVersion 

15from couchers.models.notifications import NotificationTopicAction 

16from couchers.moderation.utils import create_moderation 

17from couchers.notifications.notify import PendingNotification, notify_many 

18from couchers.proto import discussions_pb2, discussions_pb2_grpc, notification_data_pb2 

19from couchers.proto.internal import jobs_pb2 

20from couchers.servicers.api import user_model_to_pb 

21from couchers.servicers.threads import thread_to_pb 

22from couchers.sql import users_visible_to_each_other, where_moderated_content_visible 

23from couchers.utils import Timestamp_from_datetime, now 

24 

25logger = logging.getLogger(__name__) 

26 

27MAX_PAGE_SIZE = 25 

28 

29 

30def discussion_to_pb(session: Session, discussion: Discussion, context: CouchersContext) -> discussions_pb2.Discussion: 

31 owner_community_id = None 

32 owner_group_id = None 

33 if discussion.owner_cluster.is_official_cluster: 

34 owner_community_id = discussion.owner_cluster.parent_node_id 

35 else: 

36 owner_group_id = discussion.owner_cluster.id 

37 

38 if discussion.deleted is not None: 

39 return discussions_pb2.Discussion( 

40 discussion_id=discussion.id, 

41 slug=discussion.slug, 

42 deleted=True, 

43 owner_community_id=owner_community_id, 

44 owner_group_id=owner_group_id, 

45 owner_title=discussion.owner_cluster.name, 

46 thread=thread_to_pb(session, context, discussion.thread_id), 

47 ) 

48 

49 can_moderate = can_moderate_node(session, context.user_id, discussion.owner_cluster.parent_node_id) 

50 

51 return discussions_pb2.Discussion( 

52 discussion_id=discussion.id, 

53 slug=discussion.slug, 

54 created=Timestamp_from_datetime(discussion.created), 

55 creator_user_id=discussion.creator_user_id, 

56 owner_community_id=owner_community_id, 

57 owner_group_id=owner_group_id, 

58 owner_title=discussion.owner_cluster.name, 

59 title=discussion.title, 

60 content=discussion.content, 

61 thread=thread_to_pb(session, context, discussion.thread_id), 

62 can_moderate=can_moderate, 

63 can_edit=(context.user_id == discussion.creator_user_id), 

64 last_edited=Timestamp_from_datetime(discussion.last_edited) if discussion.last_edited else None, 

65 ) 

66 

67 

68def generate_create_discussion_notifications(payload: jobs_pb2.GenerateCreateDiscussionNotificationsPayload) -> None: 

69 notifications = [] 

70 with session_scope() as session: 

71 discussion = session.execute(select(Discussion).where(Discussion.id == payload.discussion_id)).scalar_one() 

72 

73 cluster = discussion.owner_cluster 

74 

75 if not cluster.is_official_cluster: 75 ↛ 76line 75 didn't jump to line 76 because the condition on line 75 was never true

76 raise NotImplementedError("Shouldn't have discussions under groups, only communities") 

77 

78 creator = aliased(User) 

79 members = ( 

80 session.execute( 

81 select(User) 

82 .join(ClusterSubscription, ClusterSubscription.user_id == User.id) 

83 .join_from(User, creator, creator.id == discussion.creator_user_id) 

84 .where(ClusterSubscription.cluster_id == cluster.id) 

85 .where(users_visible_to_each_other(self_user=User, other_user=creator)) 

86 ) 

87 .scalars() 

88 .all() 

89 ) 

90 for user in members: 

91 context = make_notification_user_context(user_id=user.id) 

92 notifications.append( 

93 PendingNotification( 

94 user_id=user.id, 

95 topic_action=NotificationTopicAction.discussion__create, 

96 key=str(payload.discussion_id), 

97 data=notification_data_pb2.DiscussionCreate( 

98 author=user_model_to_pb(discussion.creator_user, session, context), 

99 discussion=discussion_to_pb(session, discussion, context), 

100 ), 

101 moderation_state_id=discussion.moderation_state_id, 

102 ) 

103 ) 

104 

105 with session_scope() as session: 

106 notify_many(session, notifications) 

107 

108 

109class Discussions(discussions_pb2_grpc.DiscussionsServicer): 

110 def CreateDiscussion( 

111 self, request: discussions_pb2.CreateDiscussionReq, context: CouchersContext, session: Session 

112 ) -> discussions_pb2.Discussion: 

113 if not request.title: 

114 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "missing_discussion_title") 

115 if not request.content: 

116 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "missing_discussion_content") 

117 if not request.owner_community_id and not request.owner_group_id: 117 ↛ 118line 117 didn't jump to line 118 because the condition on line 117 was never true

118 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "group_or_community_not_found") 

119 

120 if request.WhichOneof("owner") == "owner_group_id": 

121 cluster = session.execute( 

122 select(Cluster).where(~Cluster.is_official_cluster).where(Cluster.id == request.owner_group_id) 

123 ).scalar_one_or_none() 

124 elif request.WhichOneof("owner") == "owner_community_id": 124 ↛ 131line 124 didn't jump to line 131 because the condition on line 124 was always true

125 cluster = session.execute( 

126 select(Cluster) 

127 .where(Cluster.parent_node_id == request.owner_community_id) 

128 .where(Cluster.is_official_cluster) 

129 ).scalar_one_or_none() 

130 

131 if not cluster: 131 ↛ 132line 131 didn't jump to line 132 because the condition on line 131 was never true

132 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "group_or_community_not_found") 

133 

134 if not cluster.small_community_features_enabled: 134 ↛ 135line 134 didn't jump to line 135 because the condition on line 134 was never true

135 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "cannot_create_discussion") 

136 

137 user = session.execute(select(User).where(User.id == context.user_id)).scalar_one() 

138 if not has_completed_profile(session, user): 

139 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "incomplete_profile_create_discussion") 

140 

141 thread = Thread() 

142 session.add(thread) 

143 session.flush() 

144 

145 discussion: Discussion | None = None 

146 

147 def create_object(moderation_state_id: int) -> int: 

148 nonlocal discussion 

149 discussion = Discussion( 

150 title=request.title, 

151 content=request.content, 

152 creator_user_id=context.user_id, 

153 owner_cluster_id=cluster.id, 

154 thread_id=thread.id, 

155 moderation_state_id=moderation_state_id, 

156 ) 

157 session.add(discussion) 

158 session.flush() 

159 return discussion.id 

160 

161 create_moderation( 

162 session=session, 

163 object_type=ModerationObjectType.discussion, 

164 object_id=create_object, 

165 creator_user_id=context.user_id, 

166 ) 

167 assert discussion is not None 

168 

169 log_event( 

170 context, 

171 session, 

172 "discussion.created", 

173 { 

174 "discussion_id": discussion.id, 

175 "cluster_id": cluster.id, 

176 "cluster_name": cluster.name, 

177 "is_official_cluster": cluster.is_official_cluster, 

178 }, 

179 ) 

180 

181 queue_job( 

182 session, 

183 job=generate_create_discussion_notifications, 

184 payload=jobs_pb2.GenerateCreateDiscussionNotificationsPayload( 

185 discussion_id=discussion.id, 

186 ), 

187 ) 

188 

189 return discussion_to_pb(session, discussion, context) 

190 

191 def GetDiscussion( 

192 self, request: discussions_pb2.GetDiscussionReq, context: CouchersContext, session: Session 

193 ) -> discussions_pb2.Discussion: 

194 discussion = session.execute( 

195 where_moderated_content_visible( 

196 select(Discussion).where(Discussion.id == request.discussion_id), 

197 context, 

198 Discussion, 

199 ) 

200 ).scalar_one_or_none() 

201 if not discussion: 

202 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "discussion_not_found") 

203 

204 return discussion_to_pb(session, discussion, context) 

205 

206 def UpdateDiscussion( 

207 self, request: discussions_pb2.UpdateDiscussionReq, context: CouchersContext, session: Session 

208 ) -> discussions_pb2.Discussion: 

209 discussion = session.execute( 

210 select(Discussion).where(Discussion.id == request.discussion_id) 

211 ).scalar_one_or_none() 

212 if not discussion: 212 ↛ 213line 212 didn't jump to line 213 because the condition on line 212 was never true

213 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "discussion_not_found") 

214 if discussion.deleted is not None: 

215 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "discussion_deleted") 

216 if context.user_id != discussion.creator_user_id: 

217 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "discussion_edit_permission_denied") 

218 

219 old_title = discussion.title 

220 old_content = discussion.content 

221 

222 if request.HasField("title"): 222 ↛ 228line 222 didn't jump to line 228 because the condition on line 222 was always true

223 new_title = request.title.value.strip() 

224 if not new_title: 224 ↛ 225line 224 didn't jump to line 225 because the condition on line 224 was never true

225 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "missing_discussion_title") 

226 discussion.title = new_title 

227 

228 if request.HasField("content"): 228 ↛ 234line 228 didn't jump to line 234 because the condition on line 228 was always true

229 new_content = request.content.value.strip() 

230 if not new_content: 230 ↛ 231line 230 didn't jump to line 231 because the condition on line 230 was never true

231 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "missing_discussion_content") 

232 discussion.content = new_content 

233 

234 title_changed = discussion.title != old_title 

235 content_changed = discussion.content != old_content 

236 

237 if not title_changed and not content_changed: 237 ↛ 238line 237 didn't jump to line 238 because the condition on line 237 was never true

238 return discussion_to_pb(session, discussion, context) 

239 

240 session.add( 

241 DiscussionVersion( 

242 discussion_id=discussion.id, 

243 editor_user_id=context.user_id, 

244 change_type=ContentChangeType.edit, 

245 old_title=old_title if title_changed else None, 

246 new_title=discussion.title if title_changed else None, 

247 old_content=old_content if content_changed else None, 

248 new_content=discussion.content if content_changed else None, 

249 ) 

250 ) 

251 

252 discussion.last_edited = now() 

253 

254 log_event( 

255 context, 

256 session, 

257 "discussion.updated", 

258 { 

259 "discussion_id": discussion.id, 

260 }, 

261 ) 

262 

263 return discussion_to_pb(session, discussion, context) 

264 

265 def DeleteDiscussion( 

266 self, request: discussions_pb2.DeleteDiscussionReq, context: CouchersContext, session: Session 

267 ) -> empty_pb2.Empty: 

268 discussion = session.execute( 

269 select(Discussion).where(Discussion.id == request.discussion_id) 

270 ).scalar_one_or_none() 

271 if not discussion: 271 ↛ 272line 271 didn't jump to line 272 because the condition on line 271 was never true

272 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "discussion_not_found") 

273 if discussion.deleted is not None: 273 ↛ 274line 273 didn't jump to line 274 because the condition on line 273 was never true

274 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "discussion_deleted") 

275 

276 if context.user_id != discussion.creator_user_id: 

277 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "discussion_delete_permission_denied") 

278 

279 session.add( 

280 DiscussionVersion( 

281 discussion_id=discussion.id, 

282 editor_user_id=context.user_id, 

283 change_type=ContentChangeType.delete, 

284 old_title=discussion.title, 

285 new_title=None, 

286 old_content=discussion.content, 

287 new_content=None, 

288 ) 

289 ) 

290 

291 discussion.deleted = now() 

292 

293 log_event( 

294 context, 

295 session, 

296 "discussion.deleted", 

297 { 

298 "discussion_id": discussion.id, 

299 }, 

300 ) 

301 

302 return empty_pb2.Empty() 

303 

304 def ListMyCommunitiesDiscussions( 

305 self, request: discussions_pb2.ListMyCommunitiesDiscussionsReq, context: CouchersContext, session: Session 

306 ) -> discussions_pb2.ListMyCommunitiesDiscussionsRes: 

307 page_size = min(MAX_PAGE_SIZE, request.page_size or MAX_PAGE_SIZE) 

308 next_page_id = int(request.page_token) if request.page_token else 2**63 - 1 

309 

310 discussions = ( 

311 session.execute( 

312 where_moderated_content_visible( 

313 select(Discussion) 

314 .join(Cluster, Cluster.id == Discussion.owner_cluster_id) 

315 .join(ClusterSubscription, ClusterSubscription.cluster_id == Cluster.id) 

316 .where(ClusterSubscription.user_id == context.user_id) 

317 .where(Cluster.is_official_cluster) 

318 .where(Cluster.small_community_features_enabled) 

319 .where(Discussion.id <= next_page_id) 

320 .order_by(Discussion.id.desc()) 

321 .limit(page_size + 1), 

322 context, 

323 Discussion, 

324 is_list_operation=True, 

325 ) 

326 ) 

327 .scalars() 

328 .all() 

329 ) 

330 

331 return discussions_pb2.ListMyCommunitiesDiscussionsRes( 

332 discussions=[discussion_to_pb(session, d, context) for d in discussions[:page_size]], 

333 next_page_token=str(discussions[-1].id) if len(discussions) > page_size else None, 

334 )