Coverage for app/backend/src/couchers/servicers/events.py: 84%

553 statements  

« prev     ^ index     » next       coverage.py v7.14.3, created at 2026-06-29 12:51 +0000

1import logging 

2from datetime import datetime, timedelta 

3from typing import Any, cast 

4 

5import grpc 

6from google.protobuf import empty_pb2 

7from psycopg.types.range import TimestamptzRange 

8from sqlalchemy import Select, select 

9from sqlalchemy.orm import Session 

10from sqlalchemy.sql import and_, func, or_, update 

11 

12from couchers.context import CouchersContext, make_notification_user_context 

13from couchers.db import can_moderate_node, get_parent_node_at_location, session_scope 

14from couchers.event_log import log_event 

15from couchers.helpers.completed_profile import has_completed_profile 

16from couchers.jobs.enqueue import queue_job 

17from couchers.models import ( 

18 AttendeeStatus, 

19 Cluster, 

20 ClusterSubscription, 

21 Event, 

22 EventCommunityInviteRequest, 

23 EventOccurrence, 

24 EventOccurrenceAttendee, 

25 EventOrganizer, 

26 EventSubscription, 

27 ModerationObjectType, 

28 Node, 

29 NodeType, 

30 Thread, 

31 Upload, 

32 User, 

33) 

34from couchers.models.notifications import NotificationTopicAction 

35from couchers.moderation.utils import create_moderation 

36from couchers.notifications.notify import notify 

37from couchers.proto import events_pb2, events_pb2_grpc, notification_data_pb2 

38from couchers.proto.internal import jobs_pb2 

39from couchers.servicers.api import user_model_to_pb 

40from couchers.servicers.blocking import is_not_visible 

41from couchers.servicers.threads import thread_to_pb 

42from couchers.sql import users_visible, where_moderated_content_visible, where_users_column_visible 

43from couchers.tasks import send_event_community_invite_request_email 

44from couchers.utils import ( 

45 Timestamp_from_datetime, 

46 create_coordinate, 

47 dt_from_millis, 

48 millis_from_dt, 

49 not_none, 

50 now, 

51 to_aware_datetime, 

52) 

53 

54logger = logging.getLogger(__name__) 

55 

56attendancestate2sql = { 

57 events_pb2.AttendanceState.ATTENDANCE_STATE_NOT_GOING: None, 

58 events_pb2.AttendanceState.ATTENDANCE_STATE_GOING: AttendeeStatus.going, 

59} 

60 

61attendancestate2api = { 

62 None: events_pb2.AttendanceState.ATTENDANCE_STATE_NOT_GOING, 

63 AttendeeStatus.going: events_pb2.AttendanceState.ATTENDANCE_STATE_GOING, 

64} 

65 

66MAX_PAGINATION_LENGTH = 25 

67 

68 

69def _is_event_owner(event: Event, user_id: int) -> bool: 

70 """ 

71 Checks whether the user can act as an owner of the event 

72 """ 

73 if event.owner_user: 

74 return event.owner_user_id == user_id 

75 # otherwise owned by a cluster 

76 return not_none(event.owner_cluster).admins.where(User.id == user_id).one_or_none() is not None 

77 

78 

79def _is_event_organizer(event: Event, user_id: int) -> bool: 

80 """ 

81 Checks whether the user is as an organizer of the event 

82 """ 

83 return event.organizers.where(EventOrganizer.user_id == user_id).one_or_none() is not None 

84 

85 

86def _can_moderate_event(session: Session, event: Event, user_id: int) -> bool: 

87 # if the event is owned by a cluster, then any moderator of that cluster can moderate this event 

88 if event.owner_cluster is not None and can_moderate_node(session, user_id, event.owner_cluster.parent_node_id): 

89 return True 

90 

91 # finally check if the user can moderate the parent node of the cluster 

92 return can_moderate_node(session, user_id, event.parent_node_id) 

93 

94 

95def _can_edit_event(session: Session, event: Event, user_id: int) -> bool: 

96 return ( 

97 _is_event_owner(event, user_id) 

98 or _is_event_organizer(event, user_id) 

99 or _can_moderate_event(session, event, user_id) 

100 ) 

101 

102 

103def event_to_pb(session: Session, occurrence: EventOccurrence, context: CouchersContext) -> events_pb2.Event: 

104 event = occurrence.event 

105 

106 next_occurrence = ( 

107 event.occurrences.where(EventOccurrence.end_time >= now()) 

108 .order_by(EventOccurrence.end_time.asc()) 

109 .limit(1) 

110 .one_or_none() 

111 ) 

112 

113 owner_community_id = None 

114 owner_group_id = None 

115 if event.owner_cluster: 

116 if event.owner_cluster.is_official_cluster: 

117 owner_community_id = event.owner_cluster.parent_node_id 

118 else: 

119 owner_group_id = event.owner_cluster.id 

120 

121 attendance = occurrence.attendances.where(EventOccurrenceAttendee.user_id == context.user_id).one_or_none() 

122 attendance_state = attendance.attendee_status if attendance else None 

123 

124 can_moderate = _can_moderate_event(session, event, context.user_id) 

125 can_edit = _can_edit_event(session, event, context.user_id) 

126 

127 going_count = session.execute( 

128 where_users_column_visible( 

129 select(func.count()) 

130 .select_from(EventOccurrenceAttendee) 

131 .where(EventOccurrenceAttendee.occurrence_id == occurrence.id) 

132 .where(EventOccurrenceAttendee.attendee_status == AttendeeStatus.going), 

133 context, 

134 EventOccurrenceAttendee.user_id, 

135 ) 

136 ).scalar_one() 

137 organizer_count = session.execute( 

138 where_users_column_visible( 

139 select(func.count()).select_from(EventOrganizer).where(EventOrganizer.event_id == event.id), 

140 context, 

141 EventOrganizer.user_id, 

142 ) 

143 ).scalar_one() 

144 subscriber_count = session.execute( 

145 where_users_column_visible( 

146 select(func.count()).select_from(EventSubscription).where(EventSubscription.event_id == event.id), 

147 context, 

148 EventSubscription.user_id, 

149 ) 

150 ).scalar_one() 

151 

152 return events_pb2.Event( 

153 event_id=occurrence.id, 

154 is_next=False if not next_occurrence else occurrence.id == next_occurrence.id, 

155 is_cancelled=occurrence.is_cancelled, 

156 is_deleted=occurrence.is_deleted, 

157 title=event.title, 

158 slug=event.slug, 

159 content=occurrence.content, 

160 photo_url=occurrence.photo.full_url if occurrence.photo else None, 

161 photo_key=occurrence.photo_key or "", 

162 online_information=( 

163 events_pb2.OnlineEventInformation( 

164 link=occurrence.link, 

165 ) 

166 if occurrence.link 

167 else None 

168 ), 

169 offline_information=( 

170 events_pb2.OfflineEventInformation( 

171 lat=not_none(occurrence.coordinates)[0], 

172 lng=not_none(occurrence.coordinates)[1], 

173 address=occurrence.address, 

174 ) 

175 if occurrence.geom 

176 else None 

177 ), 

178 created=Timestamp_from_datetime(occurrence.created), 

179 last_edited=Timestamp_from_datetime(occurrence.last_edited), 

180 creator_user_id=occurrence.creator_user_id, 

181 start_time=Timestamp_from_datetime(occurrence.start_time), 

182 end_time=Timestamp_from_datetime(occurrence.end_time), 

183 timezone=occurrence.timezone, 

184 attendance_state=attendancestate2api[attendance_state], 

185 organizer=event.organizers.where(EventOrganizer.user_id == context.user_id).one_or_none() is not None, 

186 subscriber=event.subscribers.where(EventSubscription.user_id == context.user_id).one_or_none() is not None, 

187 going_count=going_count, 

188 organizer_count=organizer_count, 

189 subscriber_count=subscriber_count, 

190 owner_user_id=event.owner_user_id, 

191 owner_community_id=owner_community_id, 

192 owner_group_id=owner_group_id, 

193 thread=thread_to_pb(session, context, event.thread_id), 

194 can_edit=can_edit, 

195 can_moderate=can_moderate, 

196 ) 

197 

198 

199def _get_event_and_occurrence_query( 

200 occurrence_id: int, 

201 include_deleted: bool, 

202 context: CouchersContext | None = None, 

203) -> Select[tuple[Event, EventOccurrence]]: 

204 query = ( 

205 select(Event, EventOccurrence) 

206 .where(EventOccurrence.id == occurrence_id) 

207 .where(EventOccurrence.event_id == Event.id) 

208 ) 

209 

210 if not include_deleted: 210 ↛ 213line 210 didn't jump to line 213 because the condition on line 210 was always true

211 query = query.where(~EventOccurrence.is_deleted) 

212 

213 if context is not None: 

214 query = where_moderated_content_visible(query, context, EventOccurrence, is_list_operation=False) 

215 

216 return query 

217 

218 

219def _get_event_and_occurrence_one( 

220 session: Session, occurrence_id: int, include_deleted: bool = False 

221) -> tuple[Event, EventOccurrence]: 

222 """For background jobs only - no visibility filtering.""" 

223 result = session.execute(_get_event_and_occurrence_query(occurrence_id, include_deleted)).one() 

224 return result._tuple() 

225 

226 

227def _get_event_and_occurrence_one_or_none( 

228 session: Session, occurrence_id: int, context: CouchersContext, include_deleted: bool = False 

229) -> tuple[Event, EventOccurrence] | None: 

230 result = session.execute( 

231 _get_event_and_occurrence_query(occurrence_id, include_deleted, context=context) 

232 ).one_or_none() 

233 return result._tuple() if result else None 

234 

235 

236def _check_occurrence_time_validity(start_time: datetime, end_time: datetime, context: CouchersContext) -> None: 

237 if start_time < now(): 

238 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "event_in_past") 

239 if end_time < start_time: 

240 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "event_ends_before_starts") 

241 if end_time - start_time > timedelta(days=7): 

242 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "event_too_long") 

243 if start_time - now() > timedelta(days=365): 

244 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "event_too_far_in_future") 

245 

246 

247def get_users_to_notify_for_new_event(session: Session, occurrence: EventOccurrence) -> tuple[list[User], int | None]: 

248 """ 

249 Returns the users to notify, as well as the community id that is being notified (None if based on geo search) 

250 """ 

251 cluster = occurrence.event.parent_node.official_cluster 

252 if occurrence.event.parent_node.node_type.value <= NodeType.region.value: 

253 logger.info("Global, macroregion, and region communities are too big for email notifications.") 

254 return [], occurrence.event.parent_node_id 

255 elif occurrence.creator_user in cluster.admins or cluster.is_leaf: 255 ↛ 258line 255 didn't jump to line 258 because the condition on line 255 was always true

256 return list(cluster.members.where(User.is_visible)), occurrence.event.parent_node_id 

257 else: 

258 max_radius = 20000 # m 

259 users = ( 

260 session.execute( 

261 select(User) 

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

263 .where(User.is_visible) 

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

265 .where(func.ST_DWithin(User.geom, occurrence.geom, max_radius / 111111)) 

266 ) 

267 .scalars() 

268 .all() 

269 ) 

270 return cast(tuple[list[User], int | None], (users, None)) 

271 

272 

273def generate_event_create_notifications(payload: jobs_pb2.GenerateEventCreateNotificationsPayload) -> None: 

274 """ 

275 Background job to generated/fan out event notifications 

276 """ 

277 # Import here to avoid circular dependency 

278 from couchers.servicers.communities import community_to_pb # noqa: PLC0415 

279 

280 logger.info(f"Fanning out notifications for event occurrence id = {payload.occurrence_id}") 

281 

282 with session_scope() as session: 

283 event, occurrence = _get_event_and_occurrence_one(session, occurrence_id=payload.occurrence_id) 

284 creator = occurrence.creator_user 

285 

286 users, node_id = get_users_to_notify_for_new_event(session, occurrence) 

287 

288 inviting_user = session.execute(select(User).where(User.id == payload.inviting_user_id)).scalar_one_or_none() 

289 

290 if not inviting_user: 290 ↛ 291line 290 didn't jump to line 291 because the condition on line 290 was never true

291 logger.error(f"Inviting user {payload.inviting_user_id} is gone while trying to send event notification?") 

292 return 

293 

294 for user in users: 

295 if is_not_visible(session, user.id, creator.id): 295 ↛ 296line 295 didn't jump to line 296 because the condition on line 295 was never true

296 continue 

297 context = make_notification_user_context(user_id=user.id) 

298 topic_action = ( 

299 NotificationTopicAction.event__create_approved 

300 if payload.approved 

301 else NotificationTopicAction.event__create_any 

302 ) 

303 notify( 

304 session, 

305 user_id=user.id, 

306 topic_action=topic_action, 

307 key=str(payload.occurrence_id), 

308 data=notification_data_pb2.EventCreate( 

309 event=event_to_pb(session, occurrence, context), 

310 inviting_user=user_model_to_pb(inviting_user, session, context), 

311 nearby=True if node_id is None else None, 

312 in_community=community_to_pb(session, event.parent_node, context) if node_id is not None else None, 

313 ), 

314 moderation_state_id=occurrence.moderation_state_id, 

315 ) 

316 

317 

318def generate_event_update_notifications(payload: jobs_pb2.GenerateEventUpdateNotificationsPayload) -> None: 

319 with session_scope() as session: 

320 event, occurrence = _get_event_and_occurrence_one(session, occurrence_id=payload.occurrence_id) 

321 

322 updating_user = session.execute(select(User).where(User.id == payload.updating_user_id)).scalar_one() 

323 

324 subscribed_user_ids = [user.id for user in event.subscribers] 

325 attending_user_ids = [user.user_id for user in occurrence.attendances] 

326 

327 for user_id in set(subscribed_user_ids + attending_user_ids): 

328 if is_not_visible(session, user_id, updating_user.id): 328 ↛ 329line 328 didn't jump to line 329 because the condition on line 328 was never true

329 continue 

330 context = make_notification_user_context(user_id=user_id) 

331 notify( 

332 session, 

333 user_id=user_id, 

334 topic_action=NotificationTopicAction.event__update, 

335 key=str(payload.occurrence_id), 

336 data=notification_data_pb2.EventUpdate( 

337 event=event_to_pb(session, occurrence, context), 

338 updating_user=user_model_to_pb(updating_user, session, context), 

339 # TODO(#9117): Remove update_str_items once known unused. 

340 updated_str_items=payload.updated_str_items, 

341 updated_enum_items=( 

342 notification_data_pb2.EventUpdateItem.ValueType(value) for value in payload.updated_enum_items 

343 ), 

344 ), 

345 moderation_state_id=occurrence.moderation_state_id, 

346 ) 

347 

348 

349def generate_event_cancel_notifications(payload: jobs_pb2.GenerateEventCancelNotificationsPayload) -> None: 

350 with session_scope() as session: 

351 event, occurrence = _get_event_and_occurrence_one(session, occurrence_id=payload.occurrence_id) 

352 

353 cancelling_user = session.execute(select(User).where(User.id == payload.cancelling_user_id)).scalar_one() 

354 

355 subscribed_user_ids = [user.id for user in event.subscribers] 

356 attending_user_ids = [user.user_id for user in occurrence.attendances] 

357 

358 for user_id in set(subscribed_user_ids + attending_user_ids): 

359 if is_not_visible(session, user_id, cancelling_user.id): 359 ↛ 360line 359 didn't jump to line 360 because the condition on line 359 was never true

360 continue 

361 context = make_notification_user_context(user_id=user_id) 

362 notify( 

363 session, 

364 user_id=user_id, 

365 topic_action=NotificationTopicAction.event__cancel, 

366 key=str(payload.occurrence_id), 

367 data=notification_data_pb2.EventCancel( 

368 event=event_to_pb(session, occurrence, context), 

369 cancelling_user=user_model_to_pb(cancelling_user, session, context), 

370 ), 

371 moderation_state_id=occurrence.moderation_state_id, 

372 ) 

373 

374 

375def generate_event_delete_notifications(payload: jobs_pb2.GenerateEventDeleteNotificationsPayload) -> None: 

376 with session_scope() as session: 

377 event, occurrence = _get_event_and_occurrence_one( 

378 session, occurrence_id=payload.occurrence_id, include_deleted=True 

379 ) 

380 

381 subscribed_user_ids = [user.id for user in event.subscribers] 

382 attending_user_ids = [user.user_id for user in occurrence.attendances] 

383 

384 for user_id in set(subscribed_user_ids + attending_user_ids): 

385 context = make_notification_user_context(user_id=user_id) 

386 notify( 

387 session, 

388 user_id=user_id, 

389 topic_action=NotificationTopicAction.event__delete, 

390 key=str(payload.occurrence_id), 

391 data=notification_data_pb2.EventDelete( 

392 event=event_to_pb(session, occurrence, context), 

393 ), 

394 moderation_state_id=occurrence.moderation_state_id, 

395 ) 

396 

397 

398class Events(events_pb2_grpc.EventsServicer): 

399 def CreateEvent( 

400 self, request: events_pb2.CreateEventReq, context: CouchersContext, session: Session 

401 ) -> events_pb2.Event: 

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

403 if not has_completed_profile(session, user): 

404 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "incomplete_profile_create_event") 

405 if not request.title: 

406 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "missing_event_title") 

407 if not request.content: 

408 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "missing_event_content") 

409 if request.HasField("online_information"): 

410 online = True 

411 geom = None 

412 address = None 

413 if not request.online_information.link: 

414 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "online_event_requires_link") 

415 link = request.online_information.link 

416 elif request.HasField("offline_information"): 416 ↛ 431line 416 didn't jump to line 431 because the condition on line 416 was always true

417 online = False 

418 # As protobuf parses a missing value as 0.0, this is not a permitted event coordinate value 

419 if not ( 

420 request.offline_information.address 

421 and request.offline_information.lat 

422 and request.offline_information.lng 

423 ): 

424 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "missing_event_address_or_location") 

425 if request.offline_information.lat == 0 and request.offline_information.lng == 0: 425 ↛ 426line 425 didn't jump to line 426 because the condition on line 425 was never true

426 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_coordinate") 

427 geom = create_coordinate(request.offline_information.lat, request.offline_information.lng) 

428 address = request.offline_information.address 

429 link = None 

430 else: 

431 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "missing_event_address_location_or_link") 

432 

433 start_time = to_aware_datetime(request.start_time) 

434 end_time = to_aware_datetime(request.end_time) 

435 

436 _check_occurrence_time_validity(start_time, end_time, context) 

437 

438 if request.parent_community_id: 

439 parent_node = session.execute( 

440 select(Node).where(Node.id == request.parent_community_id) 

441 ).scalar_one_or_none() 

442 

443 if not parent_node or not parent_node.official_cluster.small_community_features_enabled: 443 ↛ 444line 443 didn't jump to line 444 because the condition on line 443 was never true

444 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "events_not_enabled") 

445 else: 

446 if online: 

447 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "online_event_missing_parent_community") 

448 # parent community computed from geom 

449 parent_node = get_parent_node_at_location(session, not_none(geom)) 

450 

451 if not parent_node: 451 ↛ 452line 451 didn't jump to line 452 because the condition on line 451 was never true

452 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "community_not_found") 

453 

454 if ( 

455 request.photo_key 

456 and not session.execute(select(Upload).where(Upload.key == request.photo_key)).scalar_one_or_none() 

457 ): 

458 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "photo_not_found") 

459 

460 thread = Thread() 

461 session.add(thread) 

462 session.flush() 

463 

464 event = Event( 

465 title=request.title, 

466 parent_node_id=parent_node.id, 

467 owner_user_id=context.user_id, 

468 thread_id=thread.id, 

469 creator_user_id=context.user_id, 

470 ) 

471 session.add(event) 

472 session.flush() 

473 

474 occurrence: EventOccurrence | None = None 

475 

476 def create_occurrence(moderation_state_id: int) -> int: 

477 nonlocal occurrence 

478 occurrence = EventOccurrence( 

479 event_id=event.id, 

480 content=request.content, 

481 geom=geom, 

482 address=address, 

483 link=link, 

484 photo_key=request.photo_key if request.photo_key != "" else None, 

485 # timezone=timezone, 

486 during=TimestamptzRange(start_time, end_time), 

487 creator_user_id=context.user_id, 

488 moderation_state_id=moderation_state_id, 

489 ) 

490 session.add(occurrence) 

491 session.flush() 

492 return occurrence.id 

493 

494 create_moderation( 

495 session=session, 

496 object_type=ModerationObjectType.event_occurrence, 

497 object_id=create_occurrence, 

498 creator_user_id=context.user_id, 

499 ) 

500 

501 assert occurrence is not None 

502 

503 session.add( 

504 EventOrganizer( 

505 user_id=context.user_id, 

506 event_id=event.id, 

507 ) 

508 ) 

509 

510 session.add( 

511 EventSubscription( 

512 user_id=context.user_id, 

513 event_id=event.id, 

514 ) 

515 ) 

516 

517 session.add( 

518 EventOccurrenceAttendee( 

519 user_id=context.user_id, 

520 occurrence_id=occurrence.id, 

521 attendee_status=AttendeeStatus.going, 

522 ) 

523 ) 

524 

525 session.commit() 

526 

527 log_event( 

528 context, 

529 session, 

530 "event.created", 

531 { 

532 "event_id": event.id, 

533 "occurrence_id": occurrence.id, 

534 "parent_community_id": parent_node.id, 

535 "parent_community_name": parent_node.official_cluster.name, 

536 "online": online, 

537 }, 

538 ) 

539 

540 if has_completed_profile(session, user): 540 ↛ 551line 540 didn't jump to line 551 because the condition on line 540 was always true

541 queue_job( 

542 session, 

543 job=generate_event_create_notifications, 

544 payload=jobs_pb2.GenerateEventCreateNotificationsPayload( 

545 inviting_user_id=user.id, 

546 occurrence_id=occurrence.id, 

547 approved=False, 

548 ), 

549 ) 

550 

551 return event_to_pb(session, occurrence, context) 

552 

553 def ScheduleEvent( 

554 self, request: events_pb2.ScheduleEventReq, context: CouchersContext, session: Session 

555 ) -> events_pb2.Event: 

556 if not request.content: 556 ↛ 557line 556 didn't jump to line 557 because the condition on line 556 was never true

557 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "missing_event_content") 

558 if request.HasField("online_information"): 

559 geom = None 

560 address = None 

561 link = request.online_information.link 

562 elif request.HasField("offline_information"): 562 ↛ 575line 562 didn't jump to line 575 because the condition on line 562 was always true

563 if not ( 563 ↛ 568line 563 didn't jump to line 568 because the condition on line 563 was never true

564 request.offline_information.address 

565 and request.offline_information.lat 

566 and request.offline_information.lng 

567 ): 

568 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "missing_event_address_or_location") 

569 if request.offline_information.lat == 0 and request.offline_information.lng == 0: 569 ↛ 570line 569 didn't jump to line 570 because the condition on line 569 was never true

570 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_coordinate") 

571 geom = create_coordinate(request.offline_information.lat, request.offline_information.lng) 

572 address = request.offline_information.address 

573 link = None 

574 else: 

575 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "missing_event_address_location_or_link") 

576 

577 start_time = to_aware_datetime(request.start_time) 

578 end_time = to_aware_datetime(request.end_time) 

579 

580 _check_occurrence_time_validity(start_time, end_time, context) 

581 

582 res = _get_event_and_occurrence_one_or_none(session, occurrence_id=request.event_id, context=context) 

583 if not res: 583 ↛ 584line 583 didn't jump to line 584 because the condition on line 583 was never true

584 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "event_not_found") 

585 

586 event, occurrence = res 

587 

588 if not _can_edit_event(session, event, context.user_id): 588 ↛ 589line 588 didn't jump to line 589 because the condition on line 588 was never true

589 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "event_edit_permission_denied") 

590 

591 if occurrence.is_cancelled: 591 ↛ 592line 591 didn't jump to line 592 because the condition on line 591 was never true

592 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "event_cant_update_cancelled_event") 

593 

594 if ( 594 ↛ 598line 594 didn't jump to line 598 because the condition on line 594 was never true

595 request.photo_key 

596 and not session.execute(select(Upload).where(Upload.key == request.photo_key)).scalar_one_or_none() 

597 ): 

598 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "photo_not_found") 

599 

600 during = TimestamptzRange(start_time, end_time) 

601 

602 # && is the overlap operator for ranges 

603 if ( 

604 session.execute( 

605 select(EventOccurrence.id) 

606 .where(EventOccurrence.event_id == event.id) 

607 .where(EventOccurrence.during.op("&&")(during)) 

608 .limit(1) 

609 ) 

610 .scalars() 

611 .one_or_none() 

612 is not None 

613 ): 

614 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "event_cant_overlap") 

615 

616 new_occurrence: EventOccurrence | None = None 

617 

618 def create_occurrence(moderation_state_id: int) -> int: 

619 nonlocal new_occurrence 

620 new_occurrence = EventOccurrence( 

621 event_id=event.id, 

622 content=request.content, 

623 geom=geom, 

624 address=address, 

625 link=link, 

626 photo_key=request.photo_key if request.photo_key != "" else None, 

627 # timezone=timezone, 

628 during=during, 

629 creator_user_id=context.user_id, 

630 moderation_state_id=moderation_state_id, 

631 ) 

632 session.add(new_occurrence) 

633 session.flush() 

634 return new_occurrence.id 

635 

636 create_moderation( 

637 session=session, 

638 object_type=ModerationObjectType.event_occurrence, 

639 object_id=create_occurrence, 

640 creator_user_id=context.user_id, 

641 ) 

642 

643 assert new_occurrence is not None 

644 

645 session.add( 

646 EventOccurrenceAttendee( 

647 user_id=context.user_id, 

648 occurrence_id=new_occurrence.id, 

649 attendee_status=AttendeeStatus.going, 

650 ) 

651 ) 

652 

653 session.flush() 

654 

655 # TODO: notify 

656 

657 return event_to_pb(session, new_occurrence, context) 

658 

659 def UpdateEvent( 

660 self, request: events_pb2.UpdateEventReq, context: CouchersContext, session: Session 

661 ) -> events_pb2.Event: 

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

663 res = _get_event_and_occurrence_one_or_none(session, occurrence_id=request.event_id, context=context) 

664 if not res: 664 ↛ 665line 664 didn't jump to line 665 because the condition on line 664 was never true

665 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "event_not_found") 

666 

667 event, occurrence = res 

668 

669 if not _can_edit_event(session, event, context.user_id): 669 ↛ 670line 669 didn't jump to line 670 because the condition on line 669 was never true

670 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "event_edit_permission_denied") 

671 

672 # the things that were updated and need to be notified about 

673 notify_updated: list[notification_data_pb2.EventUpdateItem.ValueType] = [] 

674 

675 if occurrence.is_cancelled: 

676 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "event_cant_update_cancelled_event") 

677 

678 occurrence_update: dict[str, Any] = {"last_edited": now()} 

679 

680 if request.HasField("title"): 

681 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_TITLE) 

682 event.title = request.title.value 

683 

684 if request.HasField("content"): 

685 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_CONTENT) 

686 occurrence_update["content"] = request.content.value 

687 

688 if request.HasField("photo_key"): 688 ↛ 689line 688 didn't jump to line 689 because the condition on line 688 was never true

689 occurrence_update["photo_key"] = request.photo_key.value 

690 

691 if request.HasField("online_information"): 

692 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_LOCATION) 

693 if not request.online_information.link: 693 ↛ 694line 693 didn't jump to line 694 because the condition on line 693 was never true

694 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "online_event_requires_link") 

695 occurrence_update["link"] = request.online_information.link 

696 occurrence_update["geom"] = None 

697 occurrence_update["address"] = None 

698 elif request.HasField("offline_information"): 

699 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_LOCATION) 

700 occurrence_update["link"] = None 

701 if request.offline_information.lat == 0 and request.offline_information.lng == 0: 701 ↛ 702line 701 didn't jump to line 702 because the condition on line 701 was never true

702 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_coordinate") 

703 occurrence_update["geom"] = create_coordinate( 

704 request.offline_information.lat, request.offline_information.lng 

705 ) 

706 occurrence_update["address"] = request.offline_information.address 

707 

708 if request.HasField("start_time") or request.HasField("end_time"): 

709 if request.update_all_future: 709 ↛ 710line 709 didn't jump to line 710 because the condition on line 709 was never true

710 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "event_cant_update_all_times") 

711 if request.HasField("start_time"): 711 ↛ 715line 711 didn't jump to line 715 because the condition on line 711 was always true

712 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_START_TIME) 

713 start_time = to_aware_datetime(request.start_time) 

714 else: 

715 start_time = occurrence.start_time 

716 if request.HasField("end_time"): 

717 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_END_TIME) 

718 end_time = to_aware_datetime(request.end_time) 

719 else: 

720 end_time = occurrence.end_time 

721 

722 _check_occurrence_time_validity(start_time, end_time, context) 

723 

724 during = TimestamptzRange(start_time, end_time) 

725 

726 # && is the overlap operator for ranges 

727 if ( 

728 session.execute( 

729 select(EventOccurrence.id) 

730 .where(EventOccurrence.event_id == event.id) 

731 .where(EventOccurrence.id != occurrence.id) 

732 .where(EventOccurrence.during.op("&&")(during)) 

733 .limit(1) 

734 ) 

735 .scalars() 

736 .one_or_none() 

737 is not None 

738 ): 

739 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "event_cant_overlap") 

740 

741 occurrence_update["during"] = during 

742 

743 # TODO 

744 # if request.HasField("timezone"): 

745 # occurrence_update["timezone"] = request.timezone 

746 

747 # allow editing any event which hasn't ended more than 24 hours before now 

748 # when editing all future events, we edit all which have not yet ended 

749 

750 cutoff_time = now() - timedelta(hours=24) 

751 if request.update_all_future: 

752 session.execute( 

753 update(EventOccurrence) 

754 .where(EventOccurrence.end_time >= cutoff_time) 

755 .where(EventOccurrence.start_time >= occurrence.start_time) 

756 .values(occurrence_update) 

757 .execution_options(synchronize_session=False) 

758 ) 

759 else: 

760 if occurrence.end_time < cutoff_time: 760 ↛ 761line 760 didn't jump to line 761 because the condition on line 760 was never true

761 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "event_cant_update_old_event") 

762 session.execute( 

763 update(EventOccurrence) 

764 .where(EventOccurrence.end_time >= cutoff_time) 

765 .where(EventOccurrence.id == occurrence.id) 

766 .values(occurrence_update) 

767 .execution_options(synchronize_session=False) 

768 ) 

769 

770 session.flush() 

771 

772 if notify_updated: 

773 items_str = ",".join(notification_data_pb2.EventUpdateItem.Name(item) for item in notify_updated) 

774 if request.should_notify: 

775 logger.info(f"Items {items_str} updated in event {event.id=}, notifying") 

776 

777 queue_job( 

778 session, 

779 job=generate_event_update_notifications, 

780 payload=jobs_pb2.GenerateEventUpdateNotificationsPayload( 

781 updating_user_id=user.id, 

782 occurrence_id=occurrence.id, 

783 updated_enum_items=notify_updated, 

784 ), 

785 ) 

786 else: 

787 logger.info(f"Items {items_str} updated in event {event.id=}, but skipping notifications") 

788 

789 # since we have synchronize_session=False, we have to refresh the object 

790 session.refresh(occurrence) 

791 

792 return event_to_pb(session, occurrence, context) 

793 

794 def GetEvent(self, request: events_pb2.GetEventReq, context: CouchersContext, session: Session) -> events_pb2.Event: 

795 query = select(EventOccurrence).where(EventOccurrence.id == request.event_id).where(~EventOccurrence.is_deleted) 

796 query = where_moderated_content_visible(query, context, EventOccurrence, is_list_operation=False) 

797 occurrence = session.execute(query).scalar_one_or_none() 

798 

799 if not occurrence: 

800 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "event_not_found") 

801 

802 return event_to_pb(session, occurrence, context) 

803 

804 def CancelEvent( 

805 self, request: events_pb2.CancelEventReq, context: CouchersContext, session: Session 

806 ) -> empty_pb2.Empty: 

807 res = _get_event_and_occurrence_one_or_none(session, occurrence_id=request.event_id, context=context) 

808 if not res: 808 ↛ 809line 808 didn't jump to line 809 because the condition on line 808 was never true

809 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "event_not_found") 

810 

811 event, occurrence = res 

812 

813 if not _can_edit_event(session, event, context.user_id): 813 ↛ 814line 813 didn't jump to line 814 because the condition on line 813 was never true

814 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "event_edit_permission_denied") 

815 

816 if occurrence.end_time < now() - timedelta(hours=24): 816 ↛ 817line 816 didn't jump to line 817 because the condition on line 816 was never true

817 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "event_cant_cancel_old_event") 

818 

819 occurrence.is_cancelled = True 

820 

821 log_event(context, session, "event.cancelled", {"event_id": event.id, "occurrence_id": occurrence.id}) 

822 

823 queue_job( 

824 session, 

825 job=generate_event_cancel_notifications, 

826 payload=jobs_pb2.GenerateEventCancelNotificationsPayload( 

827 cancelling_user_id=context.user_id, 

828 occurrence_id=occurrence.id, 

829 ), 

830 ) 

831 

832 return empty_pb2.Empty() 

833 

834 def RequestCommunityInvite( 

835 self, request: events_pb2.RequestCommunityInviteReq, context: CouchersContext, session: Session 

836 ) -> empty_pb2.Empty: 

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

838 res = _get_event_and_occurrence_one_or_none(session, occurrence_id=request.event_id, context=context) 

839 if not res: 839 ↛ 840line 839 didn't jump to line 840 because the condition on line 839 was never true

840 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "event_not_found") 

841 

842 event, occurrence = res 

843 

844 if not _can_edit_event(session, event, context.user_id): 

845 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "event_edit_permission_denied") 

846 

847 if occurrence.is_cancelled: 847 ↛ 848line 847 didn't jump to line 848 because the condition on line 847 was never true

848 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "event_cant_update_cancelled_event") 

849 

850 if occurrence.end_time < now() - timedelta(hours=24): 850 ↛ 851line 850 didn't jump to line 851 because the condition on line 850 was never true

851 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "event_cant_update_old_event") 

852 

853 this_user_reqs = [req for req in occurrence.community_invite_requests if req.user_id == context.user_id] 

854 

855 if len(this_user_reqs) > 0: 

856 context.abort_with_error_code( 

857 grpc.StatusCode.FAILED_PRECONDITION, "event_community_invite_already_requested" 

858 ) 

859 

860 approved_reqs = [req for req in occurrence.community_invite_requests if req.approved] 

861 

862 if len(approved_reqs) > 0: 

863 context.abort_with_error_code( 

864 grpc.StatusCode.FAILED_PRECONDITION, "event_community_invite_already_approved" 

865 ) 

866 

867 req = EventCommunityInviteRequest( 

868 occurrence_id=request.event_id, 

869 user_id=context.user_id, 

870 ) 

871 session.add(req) 

872 session.flush() 

873 

874 send_event_community_invite_request_email(session, req) 

875 

876 return empty_pb2.Empty() 

877 

878 def ListEventOccurrences( 

879 self, request: events_pb2.ListEventOccurrencesReq, context: CouchersContext, session: Session 

880 ) -> events_pb2.ListEventOccurrencesRes: 

881 page_size = min(MAX_PAGINATION_LENGTH, request.page_size or MAX_PAGINATION_LENGTH) 

882 # the page token is a unix timestamp of where we left off 

883 page_token = dt_from_millis(int(request.page_token)) if request.page_token else now() 

884 initial_query = ( 

885 select(EventOccurrence).where(EventOccurrence.id == request.event_id).where(~EventOccurrence.is_deleted) 

886 ) 

887 initial_query = where_moderated_content_visible( 

888 initial_query, context, EventOccurrence, is_list_operation=False 

889 ) 

890 occurrence = session.execute(initial_query).scalar_one_or_none() 

891 if not occurrence: 891 ↛ 892line 891 didn't jump to line 892 because the condition on line 891 was never true

892 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "event_not_found") 

893 

894 query = ( 

895 select(EventOccurrence) 

896 .where(EventOccurrence.event_id == occurrence.event_id) 

897 .where(~EventOccurrence.is_deleted) 

898 ) 

899 query = where_moderated_content_visible(query, context, EventOccurrence, is_list_operation=True) 

900 

901 if not request.include_cancelled: 

902 query = query.where(~EventOccurrence.is_cancelled) 

903 

904 if not request.past: 904 ↛ 908line 904 didn't jump to line 908 because the condition on line 904 was always true

905 cutoff = page_token - timedelta(seconds=1) 

906 query = query.where(EventOccurrence.end_time > cutoff).order_by(EventOccurrence.start_time.asc()) 

907 else: 

908 cutoff = page_token + timedelta(seconds=1) 

909 query = query.where(EventOccurrence.end_time < cutoff).order_by(EventOccurrence.start_time.desc()) 

910 

911 query = query.limit(page_size + 1) 

912 occurrences = session.execute(query).scalars().all() 

913 

914 return events_pb2.ListEventOccurrencesRes( 

915 events=[event_to_pb(session, occurrence, context) for occurrence in occurrences[:page_size]], 

916 next_page_token=str(millis_from_dt(occurrences[-1].end_time)) if len(occurrences) > page_size else None, 

917 ) 

918 

919 def ListEventAttendees( 

920 self, request: events_pb2.ListEventAttendeesReq, context: CouchersContext, session: Session 

921 ) -> events_pb2.ListEventAttendeesRes: 

922 page_size = min(MAX_PAGINATION_LENGTH, request.page_size or MAX_PAGINATION_LENGTH) 

923 next_user_id = int(request.page_token) if request.page_token else 0 

924 occurrence = session.execute( 

925 where_moderated_content_visible( 

926 select(EventOccurrence) 

927 .where(EventOccurrence.id == request.event_id) 

928 .where(~EventOccurrence.is_deleted), 

929 context, 

930 EventOccurrence, 

931 is_list_operation=False, 

932 ) 

933 ).scalar_one_or_none() 

934 if not occurrence: 934 ↛ 935line 934 didn't jump to line 935 because the condition on line 934 was never true

935 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "event_not_found") 

936 attendees = ( 

937 session.execute( 

938 where_users_column_visible( 

939 select(EventOccurrenceAttendee) 

940 .where(EventOccurrenceAttendee.occurrence_id == occurrence.id) 

941 .where(EventOccurrenceAttendee.user_id >= next_user_id) 

942 .order_by(EventOccurrenceAttendee.user_id) 

943 .limit(page_size + 1), 

944 context, 

945 EventOccurrenceAttendee.user_id, 

946 ) 

947 ) 

948 .scalars() 

949 .all() 

950 ) 

951 return events_pb2.ListEventAttendeesRes( 

952 attendee_user_ids=[attendee.user_id for attendee in attendees[:page_size]], 

953 next_page_token=str(attendees[-1].user_id) if len(attendees) > page_size else None, 

954 ) 

955 

956 def ListEventSubscribers( 

957 self, request: events_pb2.ListEventSubscribersReq, context: CouchersContext, session: Session 

958 ) -> events_pb2.ListEventSubscribersRes: 

959 page_size = min(MAX_PAGINATION_LENGTH, request.page_size or MAX_PAGINATION_LENGTH) 

960 next_user_id = int(request.page_token) if request.page_token else 0 

961 res = _get_event_and_occurrence_one_or_none(session, occurrence_id=request.event_id, context=context) 

962 if not res: 962 ↛ 963line 962 didn't jump to line 963 because the condition on line 962 was never true

963 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "event_not_found") 

964 event, occurrence = res 

965 subscribers = ( 

966 session.execute( 

967 where_users_column_visible( 

968 select(EventSubscription) 

969 .where(EventSubscription.event_id == event.id) 

970 .where(EventSubscription.user_id >= next_user_id) 

971 .order_by(EventSubscription.user_id) 

972 .limit(page_size + 1), 

973 context, 

974 EventSubscription.user_id, 

975 ) 

976 ) 

977 .scalars() 

978 .all() 

979 ) 

980 return events_pb2.ListEventSubscribersRes( 

981 subscriber_user_ids=[subscriber.user_id for subscriber in subscribers[:page_size]], 

982 next_page_token=str(subscribers[-1].user_id) if len(subscribers) > page_size else None, 

983 ) 

984 

985 def ListEventOrganizers( 

986 self, request: events_pb2.ListEventOrganizersReq, context: CouchersContext, session: Session 

987 ) -> events_pb2.ListEventOrganizersRes: 

988 page_size = min(MAX_PAGINATION_LENGTH, request.page_size or MAX_PAGINATION_LENGTH) 

989 next_user_id = int(request.page_token) if request.page_token else 0 

990 res = _get_event_and_occurrence_one_or_none(session, occurrence_id=request.event_id, context=context) 

991 if not res: 991 ↛ 992line 991 didn't jump to line 992 because the condition on line 991 was never true

992 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "event_not_found") 

993 event, occurrence = res 

994 organizers = ( 

995 session.execute( 

996 where_users_column_visible( 

997 select(EventOrganizer) 

998 .where(EventOrganizer.event_id == event.id) 

999 .where(EventOrganizer.user_id >= next_user_id) 

1000 .order_by(EventOrganizer.user_id) 

1001 .limit(page_size + 1), 

1002 context, 

1003 EventOrganizer.user_id, 

1004 ) 

1005 ) 

1006 .scalars() 

1007 .all() 

1008 ) 

1009 return events_pb2.ListEventOrganizersRes( 

1010 organizer_user_ids=[organizer.user_id for organizer in organizers[:page_size]], 

1011 next_page_token=str(organizers[-1].user_id) if len(organizers) > page_size else None, 

1012 ) 

1013 

1014 def TransferEvent( 

1015 self, request: events_pb2.TransferEventReq, context: CouchersContext, session: Session 

1016 ) -> events_pb2.Event: 

1017 res = _get_event_and_occurrence_one_or_none(session, occurrence_id=request.event_id, context=context) 

1018 if not res: 1018 ↛ 1019line 1018 didn't jump to line 1019 because the condition on line 1018 was never true

1019 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "event_not_found") 

1020 

1021 event, occurrence = res 

1022 

1023 if not _can_edit_event(session, event, context.user_id): 

1024 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "event_transfer_permission_denied") 

1025 

1026 if occurrence.is_cancelled: 

1027 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "event_cant_update_cancelled_event") 

1028 

1029 if occurrence.end_time < now() - timedelta(hours=24): 1029 ↛ 1030line 1029 didn't jump to line 1030 because the condition on line 1029 was never true

1030 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "event_cant_update_old_event") 

1031 

1032 if request.WhichOneof("new_owner") == "new_owner_group_id": 

1033 cluster = session.execute( 

1034 select(Cluster).where(~Cluster.is_official_cluster).where(Cluster.id == request.new_owner_group_id) 

1035 ).scalar_one_or_none() 

1036 elif request.WhichOneof("new_owner") == "new_owner_community_id": 1036 ↛ 1043line 1036 didn't jump to line 1043 because the condition on line 1036 was always true

1037 cluster = session.execute( 

1038 select(Cluster) 

1039 .where(Cluster.parent_node_id == request.new_owner_community_id) 

1040 .where(Cluster.is_official_cluster) 

1041 ).scalar_one_or_none() 

1042 

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

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

1045 

1046 event.owner_user = None 

1047 event.owner_cluster = cluster 

1048 

1049 session.commit() 

1050 return event_to_pb(session, occurrence, context) 

1051 

1052 def SetEventSubscription( 

1053 self, request: events_pb2.SetEventSubscriptionReq, context: CouchersContext, session: Session 

1054 ) -> events_pb2.Event: 

1055 res = _get_event_and_occurrence_one_or_none(session, occurrence_id=request.event_id, context=context) 

1056 if not res: 1056 ↛ 1057line 1056 didn't jump to line 1057 because the condition on line 1056 was never true

1057 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "event_not_found") 

1058 

1059 event, occurrence = res 

1060 

1061 if occurrence.is_cancelled: 

1062 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "event_cant_update_cancelled_event") 

1063 

1064 if occurrence.end_time < now() - timedelta(hours=24): 1064 ↛ 1065line 1064 didn't jump to line 1065 because the condition on line 1064 was never true

1065 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "event_cant_update_old_event") 

1066 

1067 current_subscription = session.execute( 

1068 select(EventSubscription) 

1069 .where(EventSubscription.user_id == context.user_id) 

1070 .where(EventSubscription.event_id == event.id) 

1071 ).scalar_one_or_none() 

1072 

1073 # if not subscribed, subscribe 

1074 if request.subscribe and not current_subscription: 

1075 session.add(EventSubscription(user_id=context.user_id, event_id=event.id)) 

1076 

1077 # if subscribed but unsubbing, remove subscription 

1078 if not request.subscribe and current_subscription: 

1079 session.delete(current_subscription) 

1080 

1081 session.flush() 

1082 

1083 log_event( 

1084 context, 

1085 session, 

1086 "event.subscription_set", 

1087 {"event_id": event.id, "occurrence_id": occurrence.id, "subscribed": request.subscribe}, 

1088 ) 

1089 

1090 return event_to_pb(session, occurrence, context) 

1091 

1092 def SetEventAttendance( 

1093 self, request: events_pb2.SetEventAttendanceReq, context: CouchersContext, session: Session 

1094 ) -> events_pb2.Event: 

1095 occurrence = session.execute( 

1096 where_moderated_content_visible( 

1097 select(EventOccurrence) 

1098 .where(EventOccurrence.id == request.event_id) 

1099 .where(~EventOccurrence.is_deleted), 

1100 context, 

1101 EventOccurrence, 

1102 is_list_operation=False, 

1103 ) 

1104 ).scalar_one_or_none() 

1105 

1106 if not occurrence: 1106 ↛ 1107line 1106 didn't jump to line 1107 because the condition on line 1106 was never true

1107 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "event_not_found") 

1108 

1109 if occurrence.is_cancelled: 

1110 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "event_cant_update_cancelled_event") 

1111 

1112 if occurrence.end_time < now() - timedelta(hours=24): 1112 ↛ 1113line 1112 didn't jump to line 1113 because the condition on line 1112 was never true

1113 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "event_cant_update_old_event") 

1114 

1115 current_attendance = session.execute( 

1116 select(EventOccurrenceAttendee) 

1117 .where(EventOccurrenceAttendee.user_id == context.user_id) 

1118 .where(EventOccurrenceAttendee.occurrence_id == occurrence.id) 

1119 ).scalar_one_or_none() 

1120 

1121 if request.attendance_state == events_pb2.ATTENDANCE_STATE_NOT_GOING: 

1122 if current_attendance: 1122 ↛ 1137line 1122 didn't jump to line 1137 because the condition on line 1122 was always true

1123 session.delete(current_attendance) 

1124 # if unset/not going, nothing to do! 

1125 else: 

1126 if current_attendance: 1126 ↛ 1127line 1126 didn't jump to line 1127 because the condition on line 1126 was never true

1127 current_attendance.attendee_status = attendancestate2sql[request.attendance_state] # type: ignore[assignment] 

1128 else: 

1129 # create new 

1130 attendance = EventOccurrenceAttendee( 

1131 user_id=context.user_id, 

1132 occurrence_id=occurrence.id, 

1133 attendee_status=not_none(attendancestate2sql[request.attendance_state]), 

1134 ) 

1135 session.add(attendance) 

1136 

1137 session.flush() 

1138 

1139 log_event( 

1140 context, 

1141 session, 

1142 "event.attendance_set", 

1143 {"occurrence_id": occurrence.id, "attendance_state": request.attendance_state}, 

1144 ) 

1145 

1146 return event_to_pb(session, occurrence, context) 

1147 

1148 def ListMyEvents( 

1149 self, request: events_pb2.ListMyEventsReq, context: CouchersContext, session: Session 

1150 ) -> events_pb2.ListMyEventsRes: 

1151 page_size = min(MAX_PAGINATION_LENGTH, request.page_size or MAX_PAGINATION_LENGTH) 

1152 # the page token is a unix timestamp of where we left off 

1153 page_token = ( 

1154 dt_from_millis(int(request.page_token)) if request.page_token and not request.page_number else now() 

1155 ) 

1156 # the page number is the page number we are on 

1157 page_number = request.page_number or 1 

1158 # Calculate the offset for pagination 

1159 offset = (page_number - 1) * page_size 

1160 query = ( 

1161 select(EventOccurrence).join(Event, Event.id == EventOccurrence.event_id).where(~EventOccurrence.is_deleted) 

1162 ) 

1163 query = where_moderated_content_visible(query, context, EventOccurrence, is_list_operation=True) 

1164 

1165 include_all = not (request.subscribed or request.attending or request.organizing or request.my_communities) 

1166 include_subscribed = request.subscribed or include_all 

1167 include_organizing = request.organizing or include_all 

1168 include_attending = request.attending or include_all 

1169 include_my_communities = request.my_communities or include_all 

1170 

1171 if include_attending and request.exclude_attending: 

1172 context.abort_with_error_code( 

1173 grpc.StatusCode.INVALID_ARGUMENT, "cannot_combine_attending_and_exclude_attending" 

1174 ) 

1175 

1176 where_ = [] 

1177 

1178 if include_subscribed: 

1179 query = query.outerjoin( 

1180 EventSubscription, 

1181 and_(EventSubscription.event_id == Event.id, EventSubscription.user_id == context.user_id), 

1182 ) 

1183 where_.append(EventSubscription.user_id != None) 

1184 if include_organizing: 

1185 query = query.outerjoin( 

1186 EventOrganizer, and_(EventOrganizer.event_id == Event.id, EventOrganizer.user_id == context.user_id) 

1187 ) 

1188 where_.append(EventOrganizer.user_id != None) 

1189 if include_attending or request.exclude_attending: 

1190 query = query.outerjoin( 

1191 EventOccurrenceAttendee, 

1192 and_( 

1193 EventOccurrenceAttendee.occurrence_id == EventOccurrence.id, 

1194 EventOccurrenceAttendee.user_id == context.user_id, 

1195 ), 

1196 ) 

1197 if include_attending: 

1198 where_.append(EventOccurrenceAttendee.user_id != None) 

1199 elif request.exclude_attending: 1199 ↛ 1206line 1199 didn't jump to line 1206 because the condition on line 1199 was always true

1200 if not include_organizing: 1200 ↛ 1205line 1200 didn't jump to line 1205 because the condition on line 1200 was always true

1201 query = query.outerjoin( 

1202 EventOrganizer, 

1203 and_(EventOrganizer.event_id == Event.id, EventOrganizer.user_id == context.user_id), 

1204 ) 

1205 query = query.where(EventOccurrenceAttendee.user_id == None, EventOrganizer.user_id == None) 

1206 if include_my_communities: 

1207 my_communities = ( 

1208 session.execute( 

1209 select(Node.id) 

1210 .join(Cluster, Cluster.parent_node_id == Node.id) 

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

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

1213 .where(Cluster.is_official_cluster) 

1214 .order_by(Node.id) 

1215 .limit(100000) 

1216 ) 

1217 .scalars() 

1218 .all() 

1219 ) 

1220 where_.append(Event.parent_node_id.in_(my_communities)) 

1221 

1222 query = query.where(or_(*where_)) 

1223 

1224 if request.my_communities_exclude_global: 

1225 query = query.join(Node, Node.id == Event.parent_node_id).where(Node.node_type > NodeType.region) 

1226 

1227 if not request.include_cancelled: 

1228 query = query.where(~EventOccurrence.is_cancelled) 

1229 

1230 if not request.past: 1230 ↛ 1234line 1230 didn't jump to line 1234 because the condition on line 1230 was always true

1231 cutoff = page_token - timedelta(seconds=1) 

1232 query = query.where(EventOccurrence.end_time > cutoff).order_by(EventOccurrence.start_time.asc()) 

1233 else: 

1234 cutoff = page_token + timedelta(seconds=1) 

1235 query = query.where(EventOccurrence.end_time < cutoff).order_by(EventOccurrence.start_time.desc()) 

1236 # Count the total number of items for pagination 

1237 total_items = session.execute(select(func.count()).select_from(query.subquery())).scalar() 

1238 # Apply pagination by page number 

1239 query = query.offset(offset).limit(page_size) if request.page_number else query.limit(page_size + 1) 

1240 occurrences = session.execute(query).scalars().all() 

1241 

1242 return events_pb2.ListMyEventsRes( 

1243 events=[event_to_pb(session, occurrence, context) for occurrence in occurrences[:page_size]], 

1244 next_page_token=str(millis_from_dt(occurrences[-1].end_time)) if len(occurrences) > page_size else None, 

1245 total_items=total_items, 

1246 ) 

1247 

1248 def ListAllEvents( 

1249 self, request: events_pb2.ListAllEventsReq, context: CouchersContext, session: Session 

1250 ) -> events_pb2.ListAllEventsRes: 

1251 page_size = min(MAX_PAGINATION_LENGTH, request.page_size or MAX_PAGINATION_LENGTH) 

1252 # the page token is a unix timestamp of where we left off 

1253 page_token = dt_from_millis(int(request.page_token)) if request.page_token else now() 

1254 

1255 query = select(EventOccurrence).where(~EventOccurrence.is_deleted) 

1256 query = where_moderated_content_visible(query, context, EventOccurrence, is_list_operation=True) 

1257 

1258 if not request.include_cancelled: 1258 ↛ 1261line 1258 didn't jump to line 1261 because the condition on line 1258 was always true

1259 query = query.where(~EventOccurrence.is_cancelled) 

1260 

1261 if not request.past: 

1262 cutoff = page_token - timedelta(seconds=1) 

1263 query = query.where(EventOccurrence.end_time > cutoff).order_by(EventOccurrence.start_time.asc()) 

1264 else: 

1265 cutoff = page_token + timedelta(seconds=1) 

1266 query = query.where(EventOccurrence.end_time < cutoff).order_by(EventOccurrence.start_time.desc()) 

1267 

1268 query = query.limit(page_size + 1) 

1269 occurrences = session.execute(query).scalars().all() 

1270 

1271 return events_pb2.ListAllEventsRes( 

1272 events=[event_to_pb(session, occurrence, context) for occurrence in occurrences[:page_size]], 

1273 next_page_token=str(millis_from_dt(occurrences[-1].end_time)) if len(occurrences) > page_size else None, 

1274 ) 

1275 

1276 def InviteEventOrganizer( 

1277 self, request: events_pb2.InviteEventOrganizerReq, context: CouchersContext, session: Session 

1278 ) -> empty_pb2.Empty: 

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

1280 res = _get_event_and_occurrence_one_or_none(session, occurrence_id=request.event_id, context=context) 

1281 if not res: 1281 ↛ 1282line 1281 didn't jump to line 1282 because the condition on line 1281 was never true

1282 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "event_not_found") 

1283 

1284 event, occurrence = res 

1285 

1286 if not _can_edit_event(session, event, context.user_id): 

1287 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "event_edit_permission_denied") 

1288 

1289 if occurrence.is_cancelled: 

1290 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "event_cant_update_cancelled_event") 

1291 

1292 if occurrence.end_time < now() - timedelta(hours=24): 1292 ↛ 1293line 1292 didn't jump to line 1293 because the condition on line 1292 was never true

1293 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "event_cant_update_old_event") 

1294 

1295 if not session.execute( 1295 ↛ 1298line 1295 didn't jump to line 1298 because the condition on line 1295 was never true

1296 select(User).where(users_visible(context)).where(User.id == request.user_id) 

1297 ).scalar_one_or_none(): 

1298 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found") 

1299 

1300 session.add( 

1301 EventOrganizer( 

1302 user_id=request.user_id, 

1303 event_id=event.id, 

1304 ) 

1305 ) 

1306 session.flush() 

1307 

1308 other_user_context = make_notification_user_context(user_id=request.user_id) 

1309 

1310 notify( 

1311 session, 

1312 user_id=request.user_id, 

1313 topic_action=NotificationTopicAction.event__invite_organizer, 

1314 key=str(event.id), 

1315 data=notification_data_pb2.EventInviteOrganizer( 

1316 event=event_to_pb(session, occurrence, other_user_context), 

1317 inviting_user=user_model_to_pb(user, session, other_user_context), 

1318 ), 

1319 ) 

1320 

1321 return empty_pb2.Empty() 

1322 

1323 def RemoveEventOrganizer( 

1324 self, request: events_pb2.RemoveEventOrganizerReq, context: CouchersContext, session: Session 

1325 ) -> empty_pb2.Empty: 

1326 res = _get_event_and_occurrence_one_or_none(session, occurrence_id=request.event_id, context=context) 

1327 if not res: 1327 ↛ 1328line 1327 didn't jump to line 1328 because the condition on line 1327 was never true

1328 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "event_not_found") 

1329 

1330 event, occurrence = res 

1331 

1332 if occurrence.is_cancelled: 1332 ↛ 1333line 1332 didn't jump to line 1333 because the condition on line 1332 was never true

1333 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "event_cant_update_cancelled_event") 

1334 

1335 if occurrence.end_time < now() - timedelta(hours=24): 1335 ↛ 1336line 1335 didn't jump to line 1336 because the condition on line 1335 was never true

1336 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "event_cant_update_old_event") 

1337 

1338 # Determine which user to remove 

1339 user_id_to_remove = request.user_id.value if request.HasField("user_id") else context.user_id 

1340 

1341 # Check if the target user is the event owner (only after permission check) 

1342 if event.owner_user_id == user_id_to_remove: 

1343 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "event_cant_remove_owner_as_organizer") 

1344 

1345 # Check permissions: either an organizer removing an organizer OR you're the event owner 

1346 if not _can_edit_event(session, event, context.user_id): 

1347 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "event_edit_permission_denied") 

1348 

1349 # Find the organizer to remove 

1350 organizer_to_remove = session.execute( 

1351 select(EventOrganizer) 

1352 .where(EventOrganizer.user_id == user_id_to_remove) 

1353 .where(EventOrganizer.event_id == event.id) 

1354 ).scalar_one_or_none() 

1355 

1356 if not organizer_to_remove: 1356 ↛ 1357line 1356 didn't jump to line 1357 because the condition on line 1356 was never true

1357 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "event_not_an_organizer") 

1358 

1359 session.delete(organizer_to_remove) 

1360 

1361 return empty_pb2.Empty()