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

550 statements  

« prev     ^ index     » next       coverage.py v7.15.2, created at 2026-07-26 00:33 +0000

1import logging 

2from datetime import datetime, timedelta 

3from typing import Any, cast 

4from zoneinfo import ZoneInfo 

5 

6import grpc 

7from geoalchemy2 import WKBElement 

8from google.protobuf import empty_pb2 

9from psycopg.types.range import TimestamptzRange 

10from sqlalchemy import Select, func, select 

11from sqlalchemy.orm import Session 

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

13 

14from couchers.context import CouchersContext, make_notification_user_context 

15from couchers.db import can_moderate_node, get_parent_node_at_location, session_scope 

16from couchers.event_log import log_event 

17from couchers.helpers.completed_profile import has_completed_profile 

18from couchers.jobs.enqueue import queue_job 

19from couchers.models import ( 

20 AttendeeStatus, 

21 Cluster, 

22 ClusterSubscription, 

23 Event, 

24 EventCommunityInviteRequest, 

25 EventOccurrence, 

26 EventOccurrenceAttendee, 

27 EventOrganizer, 

28 EventSubscription, 

29 ModerationObjectType, 

30 Node, 

31 NodeType, 

32 Thread, 

33 Upload, 

34 User, 

35) 

36from couchers.models.notifications import NotificationTopicAction 

37from couchers.models.static import TimezoneArea 

38from couchers.moderation.utils import create_moderation 

39from couchers.notifications.notify import notify 

40from couchers.proto import events_pb2, events_pb2_grpc, notification_data_pb2 

41from couchers.proto.internal import jobs_pb2 

42from couchers.servicers.api import user_model_to_pb 

43from couchers.servicers.blocking import is_not_visible 

44from couchers.servicers.threads import thread_to_pb 

45from couchers.sql import users_visible, where_moderated_content_visible, where_users_column_visible 

46from couchers.tasks import send_event_community_invite_request_email 

47from couchers.utils import ( 

48 Timestamp_from_datetime, 

49 create_coordinate, 

50 datetime_to_iso8601_local, 

51 dt_from_millis, 

52 millis_from_dt, 

53 not_none, 

54 now, 

55) 

56 

57logger = logging.getLogger(__name__) 

58 

59attendancestate2sql = { 

60 events_pb2.AttendanceState.ATTENDANCE_STATE_NOT_GOING: None, 

61 events_pb2.AttendanceState.ATTENDANCE_STATE_GOING: AttendeeStatus.going, 

62} 

63 

64attendancestate2api = { 

65 None: events_pb2.AttendanceState.ATTENDANCE_STATE_NOT_GOING, 

66 AttendeeStatus.going: events_pb2.AttendanceState.ATTENDANCE_STATE_GOING, 

67} 

68 

69MAX_PAGINATION_LENGTH = 25 

70 

71 

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

73 """ 

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

75 """ 

76 if event.owner_user: 

77 return event.owner_user_id == user_id 

78 # otherwise owned by a cluster 

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

80 

81 

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

83 """ 

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

85 """ 

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

87 

88 

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

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

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

92 return True 

93 

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

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

96 

97 

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

99 return ( 

100 _is_event_owner(event, user_id) 

101 or _is_event_organizer(event, user_id) 

102 or _can_moderate_event(session, event, user_id) 

103 ) 

104 

105 

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

107 event = occurrence.event 

108 

109 next_occurrence = ( 

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

111 .order_by(EventOccurrence.end_time.asc()) 

112 .limit(1) 

113 .one_or_none() 

114 ) 

115 

116 owner_community_id = None 

117 owner_group_id = None 

118 if event.owner_cluster: 

119 if event.owner_cluster.is_official_cluster: 

120 owner_community_id = event.owner_cluster.parent_node_id 

121 else: 

122 owner_group_id = event.owner_cluster.id 

123 

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

125 attendance_state = attendance.attendee_status if attendance else None 

126 

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

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

129 

130 going_count = session.execute( 

131 where_users_column_visible( 

132 select(func.count()) 

133 .select_from(EventOccurrenceAttendee) 

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

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

136 context, 

137 EventOccurrenceAttendee.user_id, 

138 ) 

139 ).scalar_one() 

140 organizer_count = session.execute( 

141 where_users_column_visible( 

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

143 context, 

144 EventOrganizer.user_id, 

145 ) 

146 ).scalar_one() 

147 subscriber_count = session.execute( 

148 where_users_column_visible( 

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

150 context, 

151 EventSubscription.user_id, 

152 ) 

153 ).scalar_one() 

154 

155 return events_pb2.Event( 

156 event_id=occurrence.id, 

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

158 is_cancelled=occurrence.is_cancelled, 

159 is_deleted=occurrence.is_deleted, 

160 title=event.title, 

161 slug=event.slug, 

162 content=occurrence.content, 

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

164 photo_key=occurrence.photo_key or "", 

165 location=events_pb2.EventLocation( 

166 lat=occurrence.coordinates[0], lng=occurrence.coordinates[1], address=occurrence.address 

167 ), 

168 created=Timestamp_from_datetime(occurrence.created), 

169 last_edited=Timestamp_from_datetime(occurrence.last_edited), 

170 creator_user_id=occurrence.creator_user_id, 

171 start_time=Timestamp_from_datetime(occurrence.start_time), 

172 end_time=Timestamp_from_datetime(occurrence.end_time), 

173 timezone=occurrence.timezone, 

174 attendance_state=attendancestate2api[attendance_state], 

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

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

177 going_count=going_count, 

178 organizer_count=organizer_count, 

179 subscriber_count=subscriber_count, 

180 owner_user_id=event.owner_user_id, 

181 owner_community_id=owner_community_id, 

182 owner_group_id=owner_group_id, 

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

184 can_edit=can_edit, 

185 can_moderate=can_moderate, 

186 ) 

187 

188 

189def _get_event_and_occurrence_query( 

190 occurrence_id: int, 

191 include_deleted: bool, 

192 context: CouchersContext | None = None, 

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

194 query = ( 

195 select(Event, EventOccurrence) 

196 .where(EventOccurrence.id == occurrence_id) 

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

198 ) 

199 

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

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

202 

203 if context is not None: 

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

205 

206 return query 

207 

208 

209def _get_event_and_occurrence_one( 

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

211) -> tuple[Event, EventOccurrence]: 

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

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

214 return result._tuple() 

215 

216 

217def _get_event_and_occurrence_one_or_none( 

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

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

220 result = session.execute( 

221 _get_event_and_occurrence_query(occurrence_id, include_deleted, context=context) 

222 ).one_or_none() 

223 return result._tuple() if result else None 

224 

225 

226def _check_location(location: events_pb2.EventLocation | None, context: CouchersContext) -> tuple[WKBElement, str]: 

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

228 if not location or not location.address: 

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

230 if location.lat == 0 and location.lng == 0: 

231 # No events allowed on Null Island 

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

233 

234 geom = create_coordinate(location.lat, location.lng) 

235 return (geom, location.address) 

236 

237 

238def _check_timezone_at(geom: WKBElement, context: CouchersContext, session: Session) -> ZoneInfo: 

239 timezone_id = session.execute( 

240 select(TimezoneArea.tzid).where(func.ST_Contains(TimezoneArea.geom, func.ST_PointOnSurface(geom))).limit(1) 

241 ).scalar_one_or_none() 

242 if not timezone_id: 242 ↛ 243line 242 didn't jump to line 243 because the condition on line 242 was never true

243 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "event_timezone_not_found") 

244 

245 return ZoneInfo(timezone_id) 

246 

247 

248def _check_iso8601_local_datetime(value: str, timezone: ZoneInfo, context: CouchersContext) -> datetime: 

249 if not value: 249 ↛ 250line 249 didn't jump to line 250 because the condition on line 249 was never true

250 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "missing_event_start_end_datetime") 

251 

252 try: 

253 naive_datetime = datetime.fromisoformat(value) 

254 except ValueError: 

255 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_event_start_end_datetime") 

256 

257 if naive_datetime.tzinfo is not None: 257 ↛ 259line 257 didn't jump to line 259 because the condition on line 257 was never true

258 # Expected a local datetime, otherwise we have two sources of timezones. 

259 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_event_start_end_datetime") 

260 

261 return naive_datetime.replace(tzinfo=timezone).replace(second=0, microsecond=0) 

262 

263 

264def _update_datetime( 

265 new_iso8601_local: str | None, 

266 new_timezone: ZoneInfo, 

267 old_datetime: datetime, 

268 old_timezone: ZoneInfo, 

269 context: CouchersContext, 

270) -> datetime: 

271 if new_iso8601_local is None and new_timezone != old_timezone: 

272 # Local time wasn't updated, but the timezone changed so the effective datetime/timestamp may have changed. 

273 new_iso8601_local = datetime_to_iso8601_local(old_datetime.astimezone(old_timezone)) 

274 if new_iso8601_local is None: 

275 return old_datetime # No change 

276 # New effective datetime/timestamp 

277 return _check_iso8601_local_datetime(new_iso8601_local, new_timezone, context) 

278 

279 

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

281 if start_time < now(): 

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

283 if end_time < start_time: 

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

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

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

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

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

289 

290 

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

292 """ 

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

294 """ 

295 # people already attending or organizing the event don't need an invite to it 

296 not_already_involved = User.id.not_in( 

297 select(EventOccurrenceAttendee.user_id) 

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

299 .union(select(EventOrganizer.user_id).where(EventOrganizer.event_id == occurrence.event_id)) 

300 ) 

301 

302 cluster = occurrence.event.parent_node.official_cluster 

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

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

305 return [], occurrence.event.parent_node_id 

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

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

308 else: 

309 max_radius = 20000 # m 

310 users = ( 

311 session.execute( 

312 select(User) 

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

314 .where(User.is_visible) 

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

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

317 .where(not_already_involved) 

318 ) 

319 .scalars() 

320 .all() 

321 ) 

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

323 

324 

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

326 """ 

327 Background job to generated/fan out event notifications 

328 """ 

329 # Import here to avoid circular dependency 

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

331 

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

333 

334 with session_scope() as session: 

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

336 creator = occurrence.creator_user 

337 

338 users, node_id = get_users_to_notify_for_new_event(session, occurrence) 

339 

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

341 

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

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

344 return 

345 

346 for user in users: 

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

348 continue 

349 context = make_notification_user_context(user_id=user.id) 

350 topic_action = ( 

351 NotificationTopicAction.event__create_approved 

352 if payload.approved 

353 else NotificationTopicAction.event__create_any 

354 ) 

355 notify( 

356 session, 

357 user_id=user.id, 

358 topic_action=topic_action, 

359 key=str(payload.occurrence_id), 

360 data=notification_data_pb2.EventCreate( 

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

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

363 nearby=True if node_id is None else None, 

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

365 ), 

366 moderation_state_id=occurrence.moderation_state_id, 

367 ) 

368 

369 

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

371 with session_scope() as session: 

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

373 

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

375 

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

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

378 

379 for user_id in set(subscribed_user_ids + attending_user_ids): 

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

381 continue 

382 context = make_notification_user_context(user_id=user_id) 

383 notify( 

384 session, 

385 user_id=user_id, 

386 topic_action=NotificationTopicAction.event__update, 

387 key=str(payload.occurrence_id), 

388 data=notification_data_pb2.EventUpdate( 

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

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

391 updated_enum_items=( 

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

393 ), 

394 ), 

395 moderation_state_id=occurrence.moderation_state_id, 

396 ) 

397 

398 

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

400 with session_scope() as session: 

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

402 

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

404 

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

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

407 

408 for user_id in set(subscribed_user_ids + attending_user_ids): 

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

410 continue 

411 context = make_notification_user_context(user_id=user_id) 

412 notify( 

413 session, 

414 user_id=user_id, 

415 topic_action=NotificationTopicAction.event__cancel, 

416 key=str(payload.occurrence_id), 

417 data=notification_data_pb2.EventCancel( 

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

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

420 ), 

421 moderation_state_id=occurrence.moderation_state_id, 

422 ) 

423 

424 

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

426 with session_scope() as session: 

427 event, occurrence = _get_event_and_occurrence_one( 

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

429 ) 

430 

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

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

433 

434 for user_id in set(subscribed_user_ids + attending_user_ids): 

435 context = make_notification_user_context(user_id=user_id) 

436 notify( 

437 session, 

438 user_id=user_id, 

439 topic_action=NotificationTopicAction.event__delete, 

440 key=str(payload.occurrence_id), 

441 data=notification_data_pb2.EventDelete( 

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

443 ), 

444 moderation_state_id=occurrence.moderation_state_id, 

445 ) 

446 

447 

448class Events(events_pb2_grpc.EventsServicer): 

449 def CreateEvent( 

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

451 ) -> events_pb2.Event: 

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

453 if not has_completed_profile(session, user): 

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

455 if not request.title: 

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

457 if not request.content: 

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

459 

460 geom, address = _check_location(request.location if request.HasField("location") else None, context) 

461 timezone = _check_timezone_at(geom, context, session) 

462 start_datetime = _check_iso8601_local_datetime(request.start_datetime_iso8601_local, timezone, context) 

463 end_datetime = _check_iso8601_local_datetime(request.end_datetime_iso8601_local, timezone, context) 

464 _check_occurrence_time_validity(start_datetime, end_datetime, context) 

465 

466 if request.parent_community_id: 

467 parent_node = session.execute( 

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

469 ).scalar_one_or_none() 

470 

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

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

473 else: 

474 # parent community computed from geom 

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

476 

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

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

479 

480 if ( 

481 request.photo_key 

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

483 ): 

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

485 

486 thread = Thread() 

487 session.add(thread) 

488 session.flush() 

489 

490 event = Event( 

491 title=request.title, 

492 parent_node_id=parent_node.id, 

493 owner_user_id=context.user_id, 

494 thread_id=thread.id, 

495 creator_user_id=context.user_id, 

496 ) 

497 session.add(event) 

498 session.flush() 

499 

500 occurrence: EventOccurrence | None = None 

501 

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

503 nonlocal occurrence 

504 occurrence = EventOccurrence( 

505 event_id=event.id, 

506 content=request.content, 

507 geom=geom, 

508 address=address, 

509 timezone=timezone.key, 

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

511 during=TimestamptzRange(start_datetime, end_datetime), 

512 creator_user_id=context.user_id, 

513 moderation_state_id=moderation_state_id, 

514 ) 

515 session.add(occurrence) 

516 session.flush() 

517 return occurrence.id 

518 

519 create_moderation( 

520 session=session, 

521 object_type=ModerationObjectType.event_occurrence, 

522 object_id=create_occurrence, 

523 creator_user_id=context.user_id, 

524 ) 

525 

526 assert occurrence is not None 

527 

528 session.add( 

529 EventOrganizer( 

530 user_id=context.user_id, 

531 event_id=event.id, 

532 ) 

533 ) 

534 

535 session.add( 

536 EventSubscription( 

537 user_id=context.user_id, 

538 event_id=event.id, 

539 ) 

540 ) 

541 

542 session.add( 

543 EventOccurrenceAttendee( 

544 user_id=context.user_id, 

545 occurrence_id=occurrence.id, 

546 attendee_status=AttendeeStatus.going, 

547 ) 

548 ) 

549 

550 session.commit() 

551 

552 log_event( 

553 context, 

554 session, 

555 "event.created", 

556 { 

557 "event_id": event.id, 

558 "occurrence_id": occurrence.id, 

559 "parent_community_id": parent_node.id, 

560 "parent_community_name": parent_node.official_cluster.name, 

561 }, 

562 ) 

563 

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

565 queue_job( 

566 session, 

567 job=generate_event_create_notifications, 

568 payload=jobs_pb2.GenerateEventCreateNotificationsPayload( 

569 inviting_user_id=user.id, 

570 occurrence_id=occurrence.id, 

571 approved=False, 

572 ), 

573 ) 

574 

575 return event_to_pb(session, occurrence, context) 

576 

577 def ScheduleEvent( 

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

579 ) -> events_pb2.Event: 

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

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

582 

583 geom, address = _check_location(request.location if request.HasField("location") else None, context) 

584 timezone = _check_timezone_at(geom, context, session) 

585 start_datetime = _check_iso8601_local_datetime(request.start_datetime_iso8601_local, timezone, context) 

586 end_datetime = _check_iso8601_local_datetime(request.end_datetime_iso8601_local, timezone, context) 

587 _check_occurrence_time_validity(start_datetime, end_datetime, context) 

588 

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

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

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

592 

593 event, occurrence = res 

594 

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

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

597 

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

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

600 

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

602 request.photo_key 

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

604 ): 

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

606 

607 during = TimestamptzRange(start_datetime, end_datetime) 

608 

609 # && is the overlap operator for ranges 

610 if ( 

611 session.execute( 

612 select(EventOccurrence.id) 

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

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

615 .limit(1) 

616 ) 

617 .scalars() 

618 .one_or_none() 

619 is not None 

620 ): 

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

622 

623 new_occurrence: EventOccurrence | None = None 

624 

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

626 nonlocal new_occurrence 

627 new_occurrence = EventOccurrence( 

628 event_id=event.id, 

629 content=request.content, 

630 geom=geom, 

631 address=address, 

632 timezone=timezone.key, 

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

634 during=during, 

635 creator_user_id=context.user_id, 

636 moderation_state_id=moderation_state_id, 

637 ) 

638 session.add(new_occurrence) 

639 session.flush() 

640 return new_occurrence.id 

641 

642 create_moderation( 

643 session=session, 

644 object_type=ModerationObjectType.event_occurrence, 

645 object_id=create_occurrence, 

646 creator_user_id=context.user_id, 

647 ) 

648 

649 assert new_occurrence is not None 

650 

651 session.add( 

652 EventOccurrenceAttendee( 

653 user_id=context.user_id, 

654 occurrence_id=new_occurrence.id, 

655 attendee_status=AttendeeStatus.going, 

656 ) 

657 ) 

658 

659 session.flush() 

660 

661 # TODO: notify 

662 

663 return event_to_pb(session, new_occurrence, context) 

664 

665 def UpdateEvent( 

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

667 ) -> events_pb2.Event: 

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

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

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

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

672 

673 event, occurrence = res 

674 

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

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

677 

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

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

680 

681 if occurrence.is_cancelled: 

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

683 

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

685 

686 if request.HasField("title"): 

687 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_TITLE) 

688 event.title = request.title.value 

689 

690 if request.HasField("content"): 

691 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_CONTENT) 

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

693 

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

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

696 

697 old_timezone = ZoneInfo(occurrence.timezone) 

698 timezone: ZoneInfo = old_timezone 

699 if request.HasField("location"): 

700 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_LOCATION) 

701 geom, address = _check_location(request.location, context) 

702 timezone = _check_timezone_at(geom, context, session) 

703 occurrence_update["geom"] = geom 

704 occurrence_update["address"] = address 

705 occurrence_update["timezone"] = timezone.key 

706 

707 if timezone != old_timezone and request.update_all_future: 707 ↛ 709line 707 didn't jump to line 709 because the condition on line 707 was never true

708 # Not implemented: We'd need to change and recheck the datetimes on all existing occurrences 

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

710 

711 # Determine the new start/end datetimes, which may have changed explicitly or because of a timezone change 

712 start_datetime = _update_datetime( 

713 request.start_datetime_iso8601_local.value if request.HasField("start_datetime_iso8601_local") else None, 

714 timezone, 

715 old_datetime=occurrence.start_time, 

716 old_timezone=old_timezone, 

717 context=context, 

718 ) 

719 if start_datetime != occurrence.start_time: 

720 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_START_TIME) 

721 

722 end_datetime = _update_datetime( 

723 request.end_datetime_iso8601_local.value if request.HasField("end_datetime_iso8601_local") else None, 

724 timezone, 

725 old_datetime=occurrence.end_time, 

726 old_timezone=old_timezone, 

727 context=context, 

728 ) 

729 if end_datetime != occurrence.end_time: 

730 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_END_TIME) 

731 

732 if start_datetime != occurrence.start_time or end_datetime != occurrence.end_time: 

733 _check_occurrence_time_validity(start_datetime, end_datetime, context) 

734 

735 during = TimestamptzRange(start_datetime, end_datetime) 

736 

737 # && is the overlap operator for ranges 

738 if ( 

739 session.execute( 

740 select(EventOccurrence.id) 

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

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

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

744 .limit(1) 

745 ) 

746 .scalars() 

747 .one_or_none() 

748 is not None 

749 ): 

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

751 

752 occurrence_update["during"] = during 

753 

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

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

756 

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

758 if request.update_all_future: 

759 session.execute( 

760 update(EventOccurrence) 

761 .where(EventOccurrence.end_time >= cutoff_time) 

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

763 .values(occurrence_update) 

764 .execution_options(synchronize_session=False) 

765 ) 

766 else: 

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

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

769 session.execute( 

770 update(EventOccurrence) 

771 .where(EventOccurrence.end_time >= cutoff_time) 

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

773 .values(occurrence_update) 

774 .execution_options(synchronize_session=False) 

775 ) 

776 

777 session.flush() 

778 

779 if notify_updated: 

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

781 if request.should_notify: 

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

783 

784 queue_job( 

785 session, 

786 job=generate_event_update_notifications, 

787 payload=jobs_pb2.GenerateEventUpdateNotificationsPayload( 

788 updating_user_id=user.id, 

789 occurrence_id=occurrence.id, 

790 updated_enum_items=notify_updated, 

791 ), 

792 ) 

793 else: 

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

795 

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

797 session.refresh(occurrence) 

798 

799 return event_to_pb(session, occurrence, context) 

800 

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

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

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

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

805 

806 if not occurrence: 

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

808 

809 return event_to_pb(session, occurrence, context) 

810 

811 def CancelEvent( 

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

813 ) -> empty_pb2.Empty: 

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

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

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

817 

818 event, occurrence = res 

819 

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

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

822 

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

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

825 

826 occurrence.is_cancelled = True 

827 

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

829 

830 queue_job( 

831 session, 

832 job=generate_event_cancel_notifications, 

833 payload=jobs_pb2.GenerateEventCancelNotificationsPayload( 

834 cancelling_user_id=context.user_id, 

835 occurrence_id=occurrence.id, 

836 ), 

837 ) 

838 

839 return empty_pb2.Empty() 

840 

841 def RequestCommunityInvite( 

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

843 ) -> empty_pb2.Empty: 

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

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

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

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

848 

849 event, occurrence = res 

850 

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

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

853 

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

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

856 

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

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

859 

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

861 

862 if len(this_user_reqs) > 0: 

863 context.abort_with_error_code( 

864 grpc.StatusCode.FAILED_PRECONDITION, "event_community_invite_already_requested" 

865 ) 

866 

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

868 

869 if len(approved_reqs) > 0: 

870 context.abort_with_error_code( 

871 grpc.StatusCode.FAILED_PRECONDITION, "event_community_invite_already_approved" 

872 ) 

873 

874 req = EventCommunityInviteRequest( 

875 occurrence_id=request.event_id, 

876 user_id=context.user_id, 

877 ) 

878 session.add(req) 

879 session.flush() 

880 

881 send_event_community_invite_request_email(session, req) 

882 

883 return empty_pb2.Empty() 

884 

885 def ListEventOccurrences( 

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

887 ) -> events_pb2.ListEventOccurrencesRes: 

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

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

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

891 initial_query = ( 

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

893 ) 

894 initial_query = where_moderated_content_visible( 

895 initial_query, context, EventOccurrence, is_list_operation=False 

896 ) 

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

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

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

900 

901 query = ( 

902 select(EventOccurrence) 

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

904 .where(~EventOccurrence.is_deleted) 

905 ) 

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

907 

908 if not request.include_cancelled: 

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

910 

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

912 cutoff = page_token - timedelta(seconds=1) 

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

914 else: 

915 cutoff = page_token + timedelta(seconds=1) 

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

917 

918 query = query.limit(page_size + 1) 

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

920 

921 return events_pb2.ListEventOccurrencesRes( 

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

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

924 ) 

925 

926 def ListEventAttendees( 

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

928 ) -> events_pb2.ListEventAttendeesRes: 

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

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

931 occurrence = session.execute( 

932 where_moderated_content_visible( 

933 select(EventOccurrence) 

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

935 .where(~EventOccurrence.is_deleted), 

936 context, 

937 EventOccurrence, 

938 is_list_operation=False, 

939 ) 

940 ).scalar_one_or_none() 

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

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

943 attendees = ( 

944 session.execute( 

945 where_users_column_visible( 

946 select(EventOccurrenceAttendee) 

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

948 .where(EventOccurrenceAttendee.user_id >= next_user_id) 

949 .order_by(EventOccurrenceAttendee.user_id) 

950 .limit(page_size + 1), 

951 context, 

952 EventOccurrenceAttendee.user_id, 

953 ) 

954 ) 

955 .scalars() 

956 .all() 

957 ) 

958 return events_pb2.ListEventAttendeesRes( 

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

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

961 ) 

962 

963 def ListEventSubscribers( 

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

965 ) -> events_pb2.ListEventSubscribersRes: 

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

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

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

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

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

971 event, occurrence = res 

972 subscribers = ( 

973 session.execute( 

974 where_users_column_visible( 

975 select(EventSubscription) 

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

977 .where(EventSubscription.user_id >= next_user_id) 

978 .order_by(EventSubscription.user_id) 

979 .limit(page_size + 1), 

980 context, 

981 EventSubscription.user_id, 

982 ) 

983 ) 

984 .scalars() 

985 .all() 

986 ) 

987 return events_pb2.ListEventSubscribersRes( 

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

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

990 ) 

991 

992 def ListEventOrganizers( 

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

994 ) -> events_pb2.ListEventOrganizersRes: 

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

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

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

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

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

1000 event, occurrence = res 

1001 organizers = ( 

1002 session.execute( 

1003 where_users_column_visible( 

1004 select(EventOrganizer) 

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

1006 .where(EventOrganizer.user_id >= next_user_id) 

1007 .order_by(EventOrganizer.user_id) 

1008 .limit(page_size + 1), 

1009 context, 

1010 EventOrganizer.user_id, 

1011 ) 

1012 ) 

1013 .scalars() 

1014 .all() 

1015 ) 

1016 return events_pb2.ListEventOrganizersRes( 

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

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

1019 ) 

1020 

1021 def TransferEvent( 

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

1023 ) -> events_pb2.Event: 

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

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

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

1027 

1028 event, occurrence = res 

1029 

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

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

1032 

1033 if occurrence.is_cancelled: 

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

1035 

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

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

1038 

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

1040 cluster = session.execute( 

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

1042 ).scalar_one_or_none() 

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

1044 cluster = session.execute( 

1045 select(Cluster) 

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

1047 .where(Cluster.is_official_cluster) 

1048 ).scalar_one_or_none() 

1049 

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

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

1052 

1053 event.owner_user = None 

1054 event.owner_cluster = cluster 

1055 

1056 session.commit() 

1057 return event_to_pb(session, occurrence, context) 

1058 

1059 def SetEventSubscription( 

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

1061 ) -> events_pb2.Event: 

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

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

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

1065 

1066 event, occurrence = res 

1067 

1068 if occurrence.is_cancelled: 

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

1070 

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

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

1073 

1074 current_subscription = session.execute( 

1075 select(EventSubscription) 

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

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

1078 ).scalar_one_or_none() 

1079 

1080 # if not subscribed, subscribe 

1081 if request.subscribe and not current_subscription: 

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

1083 

1084 # if subscribed but unsubbing, remove subscription 

1085 if not request.subscribe and current_subscription: 

1086 session.delete(current_subscription) 

1087 

1088 session.flush() 

1089 

1090 log_event( 

1091 context, 

1092 session, 

1093 "event.subscription_set", 

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

1095 ) 

1096 

1097 return event_to_pb(session, occurrence, context) 

1098 

1099 def SetEventAttendance( 

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

1101 ) -> events_pb2.Event: 

1102 occurrence = session.execute( 

1103 where_moderated_content_visible( 

1104 select(EventOccurrence) 

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

1106 .where(~EventOccurrence.is_deleted), 

1107 context, 

1108 EventOccurrence, 

1109 is_list_operation=False, 

1110 ) 

1111 ).scalar_one_or_none() 

1112 

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

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

1115 

1116 if occurrence.is_cancelled: 

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

1118 

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

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

1121 

1122 current_attendance = session.execute( 

1123 select(EventOccurrenceAttendee) 

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

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

1126 ).scalar_one_or_none() 

1127 

1128 if request.attendance_state == events_pb2.ATTENDANCE_STATE_NOT_GOING: 

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

1130 session.delete(current_attendance) 

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

1132 else: 

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

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

1135 else: 

1136 # create new 

1137 attendance = EventOccurrenceAttendee( 

1138 user_id=context.user_id, 

1139 occurrence_id=occurrence.id, 

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

1141 ) 

1142 session.add(attendance) 

1143 

1144 session.flush() 

1145 

1146 log_event( 

1147 context, 

1148 session, 

1149 "event.attendance_set", 

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

1151 ) 

1152 

1153 return event_to_pb(session, occurrence, context) 

1154 

1155 def ListMyEvents( 

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

1157 ) -> events_pb2.ListMyEventsRes: 

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

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

1160 page_token = ( 

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

1162 ) 

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

1164 page_number = request.page_number or 1 

1165 # Calculate the offset for pagination 

1166 offset = (page_number - 1) * page_size 

1167 query = ( 

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

1169 ) 

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

1171 

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

1173 include_subscribed = request.subscribed or include_all 

1174 include_organizing = request.organizing or include_all 

1175 include_attending = request.attending or include_all 

1176 include_my_communities = request.my_communities or include_all 

1177 

1178 if include_attending and request.exclude_attending: 

1179 context.abort_with_error_code( 

1180 grpc.StatusCode.INVALID_ARGUMENT, "cannot_combine_attending_and_exclude_attending" 

1181 ) 

1182 

1183 where_ = [] 

1184 

1185 if include_subscribed: 

1186 query = query.outerjoin( 

1187 EventSubscription, 

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

1189 ) 

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

1191 if include_organizing: 

1192 query = query.outerjoin( 

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

1194 ) 

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

1196 if include_attending or request.exclude_attending: 

1197 query = query.outerjoin( 

1198 EventOccurrenceAttendee, 

1199 and_( 

1200 EventOccurrenceAttendee.occurrence_id == EventOccurrence.id, 

1201 EventOccurrenceAttendee.user_id == context.user_id, 

1202 ), 

1203 ) 

1204 if include_attending: 

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

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

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

1208 query = query.outerjoin( 

1209 EventOrganizer, 

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

1211 ) 

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

1213 if include_my_communities: 

1214 my_communities = ( 

1215 session.execute( 

1216 select(Node.id) 

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

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

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

1220 .where(Cluster.is_official_cluster) 

1221 .order_by(Node.id) 

1222 .limit(100000) 

1223 ) 

1224 .scalars() 

1225 .all() 

1226 ) 

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

1228 

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

1230 

1231 if request.my_communities_exclude_global: 

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

1233 

1234 if not request.include_cancelled: 

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

1236 

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

1238 cutoff = page_token - timedelta(seconds=1) 

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

1240 else: 

1241 cutoff = page_token + timedelta(seconds=1) 

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

1243 # Count the total number of items for pagination 

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

1245 # Apply pagination by page number 

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

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

1248 

1249 return events_pb2.ListMyEventsRes( 

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

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

1252 total_items=total_items, 

1253 ) 

1254 

1255 def ListAllEvents( 

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

1257 ) -> events_pb2.ListAllEventsRes: 

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

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

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

1261 

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

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

1264 

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

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

1267 

1268 if not request.past: 

1269 cutoff = page_token - timedelta(seconds=1) 

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

1271 else: 

1272 cutoff = page_token + timedelta(seconds=1) 

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

1274 

1275 query = query.limit(page_size + 1) 

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

1277 

1278 return events_pb2.ListAllEventsRes( 

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

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

1281 ) 

1282 

1283 def InviteEventOrganizer( 

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

1285 ) -> empty_pb2.Empty: 

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

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

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

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

1290 

1291 event, occurrence = res 

1292 

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

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

1295 

1296 if occurrence.is_cancelled: 

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

1298 

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

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

1301 

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

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

1304 ).scalar_one_or_none(): 

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

1306 

1307 session.add( 

1308 EventOrganizer( 

1309 user_id=request.user_id, 

1310 event_id=event.id, 

1311 ) 

1312 ) 

1313 session.flush() 

1314 

1315 other_user_context = make_notification_user_context(user_id=request.user_id) 

1316 

1317 notify( 

1318 session, 

1319 user_id=request.user_id, 

1320 topic_action=NotificationTopicAction.event__invite_organizer, 

1321 key=str(event.id), 

1322 data=notification_data_pb2.EventInviteOrganizer( 

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

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

1325 ), 

1326 ) 

1327 

1328 return empty_pb2.Empty() 

1329 

1330 def RemoveEventOrganizer( 

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

1332 ) -> empty_pb2.Empty: 

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

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

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

1336 

1337 event, occurrence = res 

1338 

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

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

1341 

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

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

1344 

1345 # Determine which user to remove 

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

1347 

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

1349 if event.owner_user_id == user_id_to_remove: 

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

1351 

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

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

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

1355 

1356 # Find the organizer to remove 

1357 organizer_to_remove = session.execute( 

1358 select(EventOrganizer) 

1359 .where(EventOrganizer.user_id == user_id_to_remove) 

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

1361 ).scalar_one_or_none() 

1362 

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

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

1365 

1366 session.delete(organizer_to_remove) 

1367 

1368 return empty_pb2.Empty()