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

560 statements  

« prev     ^ index     » next       coverage.py v7.15.2, created at 2026-07-26 11:50 +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.email.calendar_events import create_event_ics_calendar 

17from couchers.event_log import log_event 

18from couchers.helpers.completed_profile import has_completed_profile 

19from couchers.jobs.enqueue import queue_job 

20from couchers.models import ( 

21 AttendeeStatus, 

22 Cluster, 

23 ClusterSubscription, 

24 Event, 

25 EventCommunityInviteRequest, 

26 EventOccurrence, 

27 EventOccurrenceAttendee, 

28 EventOrganizer, 

29 EventSubscription, 

30 ModerationObjectType, 

31 Node, 

32 NodeType, 

33 Thread, 

34 Upload, 

35 User, 

36) 

37from couchers.models.notifications import NotificationTopicAction 

38from couchers.models.static import TimezoneArea 

39from couchers.moderation.utils import create_moderation 

40from couchers.notifications.notify import notify 

41from couchers.proto import events_pb2, events_pb2_grpc, notification_data_pb2 

42from couchers.proto.google.api import httpbody_pb2 

43from couchers.proto.internal import jobs_pb2 

44from couchers.servicers.api import user_model_to_pb 

45from couchers.servicers.blocking import is_not_visible 

46from couchers.servicers.threads import thread_to_pb 

47from couchers.sql import users_visible, where_moderated_content_visible, where_users_column_visible 

48from couchers.tasks import send_event_community_invite_request_email 

49from couchers.utils import ( 

50 Timestamp_from_datetime, 

51 create_coordinate, 

52 datetime_to_iso8601_local, 

53 dt_from_millis, 

54 millis_from_dt, 

55 not_none, 

56 now, 

57) 

58 

59logger = logging.getLogger(__name__) 

60 

61attendancestate2sql = { 

62 events_pb2.AttendanceState.ATTENDANCE_STATE_NOT_GOING: None, 

63 events_pb2.AttendanceState.ATTENDANCE_STATE_GOING: AttendeeStatus.going, 

64} 

65 

66attendancestate2api = { 

67 None: events_pb2.AttendanceState.ATTENDANCE_STATE_NOT_GOING, 

68 AttendeeStatus.going: events_pb2.AttendanceState.ATTENDANCE_STATE_GOING, 

69} 

70 

71MAX_PAGINATION_LENGTH = 25 

72 

73 

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

75 """ 

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

77 """ 

78 if event.owner_user: 

79 return event.owner_user_id == user_id 

80 # otherwise owned by a cluster 

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

82 

83 

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

85 """ 

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

87 """ 

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

89 

90 

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

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

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

94 return True 

95 

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

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

98 

99 

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

101 return ( 

102 _is_event_owner(event, user_id) 

103 or _is_event_organizer(event, user_id) 

104 or _can_moderate_event(session, event, user_id) 

105 ) 

106 

107 

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

109 event = occurrence.event 

110 

111 next_occurrence = ( 

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

113 .order_by(EventOccurrence.end_time.asc()) 

114 .limit(1) 

115 .one_or_none() 

116 ) 

117 

118 owner_community_id = None 

119 owner_group_id = None 

120 if event.owner_cluster: 

121 if event.owner_cluster.is_official_cluster: 

122 owner_community_id = event.owner_cluster.parent_node_id 

123 else: 

124 owner_group_id = event.owner_cluster.id 

125 

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

127 attendance_state = attendance.attendee_status if attendance else None 

128 

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

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

131 

132 going_count = session.execute( 

133 where_users_column_visible( 

134 select(func.count()) 

135 .select_from(EventOccurrenceAttendee) 

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

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

138 context, 

139 EventOccurrenceAttendee.user_id, 

140 ) 

141 ).scalar_one() 

142 organizer_count = session.execute( 

143 where_users_column_visible( 

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

145 context, 

146 EventOrganizer.user_id, 

147 ) 

148 ).scalar_one() 

149 subscriber_count = session.execute( 

150 where_users_column_visible( 

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

152 context, 

153 EventSubscription.user_id, 

154 ) 

155 ).scalar_one() 

156 

157 return events_pb2.Event( 

158 event_id=occurrence.id, 

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

160 is_cancelled=occurrence.is_cancelled, 

161 is_deleted=occurrence.is_deleted, 

162 title=event.title, 

163 slug=event.slug, 

164 content=occurrence.content, 

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

166 photo_key=occurrence.photo_key or "", 

167 location=events_pb2.EventLocation( 

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

169 ), 

170 created=Timestamp_from_datetime(occurrence.created), 

171 last_edited=Timestamp_from_datetime(occurrence.last_edited), 

172 creator_user_id=occurrence.creator_user_id, 

173 start_time=Timestamp_from_datetime(occurrence.start_time), 

174 end_time=Timestamp_from_datetime(occurrence.end_time), 

175 timezone=occurrence.timezone, 

176 attendance_state=attendancestate2api[attendance_state], 

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

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

179 going_count=going_count, 

180 organizer_count=organizer_count, 

181 subscriber_count=subscriber_count, 

182 owner_user_id=event.owner_user_id, 

183 owner_community_id=owner_community_id, 

184 owner_group_id=owner_group_id, 

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

186 can_edit=can_edit, 

187 can_moderate=can_moderate, 

188 ) 

189 

190 

191def _get_event_and_occurrence_query( 

192 occurrence_id: int, 

193 include_deleted: bool, 

194 context: CouchersContext | None = None, 

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

196 query = ( 

197 select(Event, EventOccurrence) 

198 .where(EventOccurrence.id == occurrence_id) 

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

200 ) 

201 

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

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

204 

205 if context is not None: 

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

207 

208 return query 

209 

210 

211def _get_event_and_occurrence_one( 

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

213) -> tuple[Event, EventOccurrence]: 

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

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

216 return result._tuple() 

217 

218 

219def _get_event_and_occurrence_one_or_none( 

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

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

222 result = session.execute( 

223 _get_event_and_occurrence_query(occurrence_id, include_deleted, context=context) 

224 ).one_or_none() 

225 return result._tuple() if result else None 

226 

227 

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

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

230 if not location or not location.address: 

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

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

233 # No events allowed on Null Island 

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

235 

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

237 return (geom, location.address) 

238 

239 

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

241 timezone_id = session.execute( 

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

243 ).scalar_one_or_none() 

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

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

246 

247 return ZoneInfo(timezone_id) 

248 

249 

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

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

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

253 

254 try: 

255 naive_datetime = datetime.fromisoformat(value) 

256 except ValueError: 

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

258 

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

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

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

262 

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

264 

265 

266def _update_datetime( 

267 new_iso8601_local: str | None, 

268 new_timezone: ZoneInfo, 

269 old_datetime: datetime, 

270 old_timezone: ZoneInfo, 

271 context: CouchersContext, 

272) -> datetime: 

273 if new_iso8601_local is None and new_timezone != old_timezone: 

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

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

276 if new_iso8601_local is None: 

277 return old_datetime # No change 

278 # New effective datetime/timestamp 

279 return _check_iso8601_local_datetime(new_iso8601_local, new_timezone, context) 

280 

281 

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

283 if start_time < now(): 

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

285 if end_time < start_time: 

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

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

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

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

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

291 

292 

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

294 """ 

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

296 """ 

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

298 not_already_involved = User.id.not_in( 

299 select(EventOccurrenceAttendee.user_id) 

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

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

302 ) 

303 

304 cluster = occurrence.event.parent_node.official_cluster 

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

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

307 return [], occurrence.event.parent_node_id 

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

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

310 else: 

311 max_radius = 20000 # m 

312 users = ( 

313 session.execute( 

314 select(User) 

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

316 .where(User.is_visible) 

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

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

319 .where(not_already_involved) 

320 ) 

321 .scalars() 

322 .all() 

323 ) 

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

325 

326 

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

328 """ 

329 Background job to generated/fan out event notifications 

330 """ 

331 # Import here to avoid circular dependency 

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

333 

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

335 

336 with session_scope() as session: 

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

338 creator = occurrence.creator_user 

339 

340 users, node_id = get_users_to_notify_for_new_event(session, occurrence) 

341 

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

343 

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

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

346 return 

347 

348 for user in users: 

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

350 continue 

351 context = make_notification_user_context(user_id=user.id) 

352 topic_action = ( 

353 NotificationTopicAction.event__create_approved 

354 if payload.approved 

355 else NotificationTopicAction.event__create_any 

356 ) 

357 notify( 

358 session, 

359 user_id=user.id, 

360 topic_action=topic_action, 

361 key=str(payload.occurrence_id), 

362 data=notification_data_pb2.EventCreate( 

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

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

365 nearby=True if node_id is None else None, 

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

367 ), 

368 moderation_state_id=occurrence.moderation_state_id, 

369 ) 

370 

371 

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

373 with session_scope() as session: 

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

375 

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

377 

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

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

380 

381 for user_id in set(subscribed_user_ids + attending_user_ids): 

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

383 continue 

384 context = make_notification_user_context(user_id=user_id) 

385 notify( 

386 session, 

387 user_id=user_id, 

388 topic_action=NotificationTopicAction.event__update, 

389 key=str(payload.occurrence_id), 

390 data=notification_data_pb2.EventUpdate( 

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

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

393 updated_enum_items=( 

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

395 ), 

396 ), 

397 moderation_state_id=occurrence.moderation_state_id, 

398 ) 

399 

400 

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

402 with session_scope() as session: 

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

404 

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

406 

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

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

409 

410 for user_id in set(subscribed_user_ids + attending_user_ids): 

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

412 continue 

413 context = make_notification_user_context(user_id=user_id) 

414 notify( 

415 session, 

416 user_id=user_id, 

417 topic_action=NotificationTopicAction.event__cancel, 

418 key=str(payload.occurrence_id), 

419 data=notification_data_pb2.EventCancel( 

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

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

422 ), 

423 moderation_state_id=occurrence.moderation_state_id, 

424 ) 

425 

426 

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

428 with session_scope() as session: 

429 event, occurrence = _get_event_and_occurrence_one( 

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

431 ) 

432 

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

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

435 

436 for user_id in set(subscribed_user_ids + attending_user_ids): 

437 context = make_notification_user_context(user_id=user_id) 

438 notify( 

439 session, 

440 user_id=user_id, 

441 topic_action=NotificationTopicAction.event__delete, 

442 key=str(payload.occurrence_id), 

443 data=notification_data_pb2.EventDelete( 

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

445 ), 

446 moderation_state_id=occurrence.moderation_state_id, 

447 ) 

448 

449 

450class Events(events_pb2_grpc.EventsServicer): 

451 def CreateEvent( 

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

453 ) -> events_pb2.Event: 

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

455 if not has_completed_profile(session, user): 

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

457 if not request.title: 

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

459 if not request.content: 

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

461 

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

463 timezone = _check_timezone_at(geom, context, session) 

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

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

466 _check_occurrence_time_validity(start_datetime, end_datetime, context) 

467 

468 if request.parent_community_id: 

469 parent_node = session.execute( 

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

471 ).scalar_one_or_none() 

472 

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

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

475 else: 

476 # parent community computed from geom 

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

478 

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

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

481 

482 if ( 

483 request.photo_key 

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

485 ): 

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

487 

488 thread = Thread() 

489 session.add(thread) 

490 session.flush() 

491 

492 event = Event( 

493 title=request.title, 

494 parent_node_id=parent_node.id, 

495 owner_user_id=context.user_id, 

496 thread_id=thread.id, 

497 creator_user_id=context.user_id, 

498 ) 

499 session.add(event) 

500 session.flush() 

501 

502 occurrence: EventOccurrence | None = None 

503 

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

505 nonlocal occurrence 

506 occurrence = EventOccurrence( 

507 event_id=event.id, 

508 content=request.content, 

509 geom=geom, 

510 address=address, 

511 timezone=timezone.key, 

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

513 during=TimestamptzRange(start_datetime, end_datetime), 

514 creator_user_id=context.user_id, 

515 moderation_state_id=moderation_state_id, 

516 ) 

517 session.add(occurrence) 

518 session.flush() 

519 return occurrence.id 

520 

521 create_moderation( 

522 session=session, 

523 object_type=ModerationObjectType.event_occurrence, 

524 object_id=create_occurrence, 

525 creator_user_id=context.user_id, 

526 ) 

527 

528 assert occurrence is not None 

529 

530 session.add( 

531 EventOrganizer( 

532 user_id=context.user_id, 

533 event_id=event.id, 

534 ) 

535 ) 

536 

537 session.add( 

538 EventSubscription( 

539 user_id=context.user_id, 

540 event_id=event.id, 

541 ) 

542 ) 

543 

544 session.add( 

545 EventOccurrenceAttendee( 

546 user_id=context.user_id, 

547 occurrence_id=occurrence.id, 

548 attendee_status=AttendeeStatus.going, 

549 ) 

550 ) 

551 

552 session.commit() 

553 

554 log_event( 

555 context, 

556 session, 

557 "event.created", 

558 { 

559 "event_id": event.id, 

560 "occurrence_id": occurrence.id, 

561 "parent_community_id": parent_node.id, 

562 "parent_community_name": parent_node.official_cluster.name, 

563 }, 

564 ) 

565 

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

567 queue_job( 

568 session, 

569 job=generate_event_create_notifications, 

570 payload=jobs_pb2.GenerateEventCreateNotificationsPayload( 

571 inviting_user_id=user.id, 

572 occurrence_id=occurrence.id, 

573 approved=False, 

574 ), 

575 ) 

576 

577 return event_to_pb(session, occurrence, context) 

578 

579 def ScheduleEvent( 

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

581 ) -> events_pb2.Event: 

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

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

584 

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

586 timezone = _check_timezone_at(geom, context, session) 

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

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

589 _check_occurrence_time_validity(start_datetime, end_datetime, context) 

590 

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

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

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

594 

595 event, occurrence = res 

596 

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

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

599 

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

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

602 

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

604 request.photo_key 

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

606 ): 

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

608 

609 during = TimestamptzRange(start_datetime, end_datetime) 

610 

611 # && is the overlap operator for ranges 

612 if ( 

613 session.execute( 

614 select(EventOccurrence.id) 

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

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

617 .limit(1) 

618 ) 

619 .scalars() 

620 .one_or_none() 

621 is not None 

622 ): 

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

624 

625 new_occurrence: EventOccurrence | None = None 

626 

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

628 nonlocal new_occurrence 

629 new_occurrence = EventOccurrence( 

630 event_id=event.id, 

631 content=request.content, 

632 geom=geom, 

633 address=address, 

634 timezone=timezone.key, 

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

636 during=during, 

637 creator_user_id=context.user_id, 

638 moderation_state_id=moderation_state_id, 

639 ) 

640 session.add(new_occurrence) 

641 session.flush() 

642 return new_occurrence.id 

643 

644 create_moderation( 

645 session=session, 

646 object_type=ModerationObjectType.event_occurrence, 

647 object_id=create_occurrence, 

648 creator_user_id=context.user_id, 

649 ) 

650 

651 assert new_occurrence is not None 

652 

653 session.add( 

654 EventOccurrenceAttendee( 

655 user_id=context.user_id, 

656 occurrence_id=new_occurrence.id, 

657 attendee_status=AttendeeStatus.going, 

658 ) 

659 ) 

660 

661 session.flush() 

662 

663 # TODO: notify 

664 

665 return event_to_pb(session, new_occurrence, context) 

666 

667 def UpdateEvent( 

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

669 ) -> events_pb2.Event: 

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

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

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

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

674 

675 event, occurrence = res 

676 

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

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

679 

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

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

682 

683 if occurrence.is_cancelled: 

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

685 

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

687 

688 if request.HasField("title"): 

689 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_TITLE) 

690 event.title = request.title.value 

691 

692 if request.HasField("content"): 

693 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_CONTENT) 

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

695 

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

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

698 

699 old_timezone = ZoneInfo(occurrence.timezone) 

700 timezone: ZoneInfo = old_timezone 

701 if request.HasField("location"): 

702 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_LOCATION) 

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

704 timezone = _check_timezone_at(geom, context, session) 

705 occurrence_update["geom"] = geom 

706 occurrence_update["address"] = address 

707 occurrence_update["timezone"] = timezone.key 

708 

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

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

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

712 

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

714 start_datetime = _update_datetime( 

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

716 timezone, 

717 old_datetime=occurrence.start_time, 

718 old_timezone=old_timezone, 

719 context=context, 

720 ) 

721 if start_datetime != occurrence.start_time: 

722 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_START_TIME) 

723 

724 end_datetime = _update_datetime( 

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

726 timezone, 

727 old_datetime=occurrence.end_time, 

728 old_timezone=old_timezone, 

729 context=context, 

730 ) 

731 if end_datetime != occurrence.end_time: 

732 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_END_TIME) 

733 

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

735 _check_occurrence_time_validity(start_datetime, end_datetime, context) 

736 

737 during = TimestamptzRange(start_datetime, end_datetime) 

738 

739 # && is the overlap operator for ranges 

740 if ( 

741 session.execute( 

742 select(EventOccurrence.id) 

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

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

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

746 .limit(1) 

747 ) 

748 .scalars() 

749 .one_or_none() 

750 is not None 

751 ): 

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

753 

754 occurrence_update["during"] = during 

755 

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

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

758 

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

760 if request.update_all_future: 

761 session.execute( 

762 update(EventOccurrence) 

763 .where(EventOccurrence.end_time >= cutoff_time) 

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

765 .values(occurrence_update) 

766 .execution_options(synchronize_session=False) 

767 ) 

768 else: 

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

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

771 session.execute( 

772 update(EventOccurrence) 

773 .where(EventOccurrence.end_time >= cutoff_time) 

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

775 .values(occurrence_update) 

776 .execution_options(synchronize_session=False) 

777 ) 

778 

779 session.flush() 

780 

781 if notify_updated: 

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

783 if request.should_notify: 

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

785 

786 queue_job( 

787 session, 

788 job=generate_event_update_notifications, 

789 payload=jobs_pb2.GenerateEventUpdateNotificationsPayload( 

790 updating_user_id=user.id, 

791 occurrence_id=occurrence.id, 

792 updated_enum_items=notify_updated, 

793 ), 

794 ) 

795 else: 

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

797 

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

799 session.refresh(occurrence) 

800 

801 return event_to_pb(session, occurrence, context) 

802 

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

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

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

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

807 

808 if not occurrence: 

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

810 

811 return event_to_pb(session, occurrence, context) 

812 

813 def CancelEvent( 

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

815 ) -> empty_pb2.Empty: 

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

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

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

819 

820 event, occurrence = res 

821 

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

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

824 

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

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

827 

828 occurrence.is_cancelled = True 

829 

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

831 

832 queue_job( 

833 session, 

834 job=generate_event_cancel_notifications, 

835 payload=jobs_pb2.GenerateEventCancelNotificationsPayload( 

836 cancelling_user_id=context.user_id, 

837 occurrence_id=occurrence.id, 

838 ), 

839 ) 

840 

841 return empty_pb2.Empty() 

842 

843 def RequestCommunityInvite( 

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

845 ) -> empty_pb2.Empty: 

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

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

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

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

850 

851 event, occurrence = res 

852 

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

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

855 

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

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

858 

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

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

861 

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

863 

864 if len(this_user_reqs) > 0: 

865 context.abort_with_error_code( 

866 grpc.StatusCode.FAILED_PRECONDITION, "event_community_invite_already_requested" 

867 ) 

868 

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

870 

871 if len(approved_reqs) > 0: 

872 context.abort_with_error_code( 

873 grpc.StatusCode.FAILED_PRECONDITION, "event_community_invite_already_approved" 

874 ) 

875 

876 req = EventCommunityInviteRequest( 

877 occurrence_id=request.event_id, 

878 user_id=context.user_id, 

879 ) 

880 session.add(req) 

881 session.flush() 

882 

883 send_event_community_invite_request_email(session, req) 

884 

885 return empty_pb2.Empty() 

886 

887 def ListEventOccurrences( 

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

889 ) -> events_pb2.ListEventOccurrencesRes: 

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

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

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

893 initial_query = ( 

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

895 ) 

896 initial_query = where_moderated_content_visible( 

897 initial_query, context, EventOccurrence, is_list_operation=False 

898 ) 

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

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

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

902 

903 query = ( 

904 select(EventOccurrence) 

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

906 .where(~EventOccurrence.is_deleted) 

907 ) 

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

909 

910 if not request.include_cancelled: 

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

912 

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

914 cutoff = page_token - timedelta(seconds=1) 

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

916 else: 

917 cutoff = page_token + timedelta(seconds=1) 

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

919 

920 query = query.limit(page_size + 1) 

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

922 

923 return events_pb2.ListEventOccurrencesRes( 

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

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

926 ) 

927 

928 def ListEventAttendees( 

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

930 ) -> events_pb2.ListEventAttendeesRes: 

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

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

933 occurrence = session.execute( 

934 where_moderated_content_visible( 

935 select(EventOccurrence) 

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

937 .where(~EventOccurrence.is_deleted), 

938 context, 

939 EventOccurrence, 

940 is_list_operation=False, 

941 ) 

942 ).scalar_one_or_none() 

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

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

945 attendees = ( 

946 session.execute( 

947 where_users_column_visible( 

948 select(EventOccurrenceAttendee) 

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

950 .where(EventOccurrenceAttendee.user_id >= next_user_id) 

951 .order_by(EventOccurrenceAttendee.user_id) 

952 .limit(page_size + 1), 

953 context, 

954 EventOccurrenceAttendee.user_id, 

955 ) 

956 ) 

957 .scalars() 

958 .all() 

959 ) 

960 return events_pb2.ListEventAttendeesRes( 

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

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

963 ) 

964 

965 def ListEventSubscribers( 

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

967 ) -> events_pb2.ListEventSubscribersRes: 

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

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

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

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

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

973 event, occurrence = res 

974 subscribers = ( 

975 session.execute( 

976 where_users_column_visible( 

977 select(EventSubscription) 

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

979 .where(EventSubscription.user_id >= next_user_id) 

980 .order_by(EventSubscription.user_id) 

981 .limit(page_size + 1), 

982 context, 

983 EventSubscription.user_id, 

984 ) 

985 ) 

986 .scalars() 

987 .all() 

988 ) 

989 return events_pb2.ListEventSubscribersRes( 

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

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

992 ) 

993 

994 def ListEventOrganizers( 

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

996 ) -> events_pb2.ListEventOrganizersRes: 

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

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

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

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

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

1002 event, occurrence = res 

1003 organizers = ( 

1004 session.execute( 

1005 where_users_column_visible( 

1006 select(EventOrganizer) 

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

1008 .where(EventOrganizer.user_id >= next_user_id) 

1009 .order_by(EventOrganizer.user_id) 

1010 .limit(page_size + 1), 

1011 context, 

1012 EventOrganizer.user_id, 

1013 ) 

1014 ) 

1015 .scalars() 

1016 .all() 

1017 ) 

1018 return events_pb2.ListEventOrganizersRes( 

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

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

1021 ) 

1022 

1023 def TransferEvent( 

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

1025 ) -> events_pb2.Event: 

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

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

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

1029 

1030 event, occurrence = res 

1031 

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

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

1034 

1035 if occurrence.is_cancelled: 

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

1037 

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

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

1040 

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

1042 cluster = session.execute( 

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

1044 ).scalar_one_or_none() 

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

1046 cluster = session.execute( 

1047 select(Cluster) 

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

1049 .where(Cluster.is_official_cluster) 

1050 ).scalar_one_or_none() 

1051 

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

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

1054 

1055 event.owner_user = None 

1056 event.owner_cluster = cluster 

1057 

1058 session.commit() 

1059 return event_to_pb(session, occurrence, context) 

1060 

1061 def SetEventSubscription( 

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

1063 ) -> events_pb2.Event: 

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

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

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

1067 

1068 event, occurrence = res 

1069 

1070 if occurrence.is_cancelled: 

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

1072 

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

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

1075 

1076 current_subscription = session.execute( 

1077 select(EventSubscription) 

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

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

1080 ).scalar_one_or_none() 

1081 

1082 # if not subscribed, subscribe 

1083 if request.subscribe and not current_subscription: 

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

1085 

1086 # if subscribed but unsubbing, remove subscription 

1087 if not request.subscribe and current_subscription: 

1088 session.delete(current_subscription) 

1089 

1090 session.flush() 

1091 

1092 log_event( 

1093 context, 

1094 session, 

1095 "event.subscription_set", 

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

1097 ) 

1098 

1099 return event_to_pb(session, occurrence, context) 

1100 

1101 def SetEventAttendance( 

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

1103 ) -> events_pb2.Event: 

1104 occurrence = session.execute( 

1105 where_moderated_content_visible( 

1106 select(EventOccurrence) 

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

1108 .where(~EventOccurrence.is_deleted), 

1109 context, 

1110 EventOccurrence, 

1111 is_list_operation=False, 

1112 ) 

1113 ).scalar_one_or_none() 

1114 

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

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

1117 

1118 if occurrence.is_cancelled: 

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

1120 

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

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

1123 

1124 current_attendance = session.execute( 

1125 select(EventOccurrenceAttendee) 

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

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

1128 ).scalar_one_or_none() 

1129 

1130 if request.attendance_state == events_pb2.ATTENDANCE_STATE_NOT_GOING: 

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

1132 session.delete(current_attendance) 

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

1134 else: 

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

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

1137 else: 

1138 # create new 

1139 attendance = EventOccurrenceAttendee( 

1140 user_id=context.user_id, 

1141 occurrence_id=occurrence.id, 

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

1143 ) 

1144 session.add(attendance) 

1145 

1146 session.flush() 

1147 

1148 log_event( 

1149 context, 

1150 session, 

1151 "event.attendance_set", 

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

1153 ) 

1154 

1155 return event_to_pb(session, occurrence, context) 

1156 

1157 def ListMyEvents( 

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

1159 ) -> events_pb2.ListMyEventsRes: 

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

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

1162 page_token = ( 

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

1164 ) 

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

1166 page_number = request.page_number or 1 

1167 # Calculate the offset for pagination 

1168 offset = (page_number - 1) * page_size 

1169 query = ( 

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

1171 ) 

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

1173 

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

1175 include_subscribed = request.subscribed or include_all 

1176 include_organizing = request.organizing or include_all 

1177 include_attending = request.attending or include_all 

1178 include_my_communities = request.my_communities or include_all 

1179 

1180 if include_attending and request.exclude_attending: 

1181 context.abort_with_error_code( 

1182 grpc.StatusCode.INVALID_ARGUMENT, "cannot_combine_attending_and_exclude_attending" 

1183 ) 

1184 

1185 where_ = [] 

1186 

1187 if include_subscribed: 

1188 query = query.outerjoin( 

1189 EventSubscription, 

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

1191 ) 

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

1193 if include_organizing: 

1194 query = query.outerjoin( 

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

1196 ) 

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

1198 if include_attending or request.exclude_attending: 

1199 query = query.outerjoin( 

1200 EventOccurrenceAttendee, 

1201 and_( 

1202 EventOccurrenceAttendee.occurrence_id == EventOccurrence.id, 

1203 EventOccurrenceAttendee.user_id == context.user_id, 

1204 ), 

1205 ) 

1206 if include_attending: 

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

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

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

1210 query = query.outerjoin( 

1211 EventOrganizer, 

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

1213 ) 

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

1215 if include_my_communities: 

1216 my_communities = ( 

1217 session.execute( 

1218 select(Node.id) 

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

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

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

1222 .where(Cluster.is_official_cluster) 

1223 .order_by(Node.id) 

1224 .limit(100000) 

1225 ) 

1226 .scalars() 

1227 .all() 

1228 ) 

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

1230 

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

1232 

1233 if request.my_communities_exclude_global: 

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

1235 

1236 if not request.include_cancelled: 

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

1238 

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

1240 cutoff = page_token - timedelta(seconds=1) 

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

1242 else: 

1243 cutoff = page_token + timedelta(seconds=1) 

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

1245 # Count the total number of items for pagination 

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

1247 # Apply pagination by page number 

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

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

1250 

1251 return events_pb2.ListMyEventsRes( 

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

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

1254 total_items=total_items, 

1255 ) 

1256 

1257 def ListAllEvents( 

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

1259 ) -> events_pb2.ListAllEventsRes: 

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

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

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

1263 

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

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

1266 

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

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

1269 

1270 if not request.past: 

1271 cutoff = page_token - timedelta(seconds=1) 

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

1273 else: 

1274 cutoff = page_token + timedelta(seconds=1) 

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

1276 

1277 query = query.limit(page_size + 1) 

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

1279 

1280 return events_pb2.ListAllEventsRes( 

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

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

1283 ) 

1284 

1285 def InviteEventOrganizer( 

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

1287 ) -> empty_pb2.Empty: 

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

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

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

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

1292 

1293 event, occurrence = res 

1294 

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

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

1297 

1298 if occurrence.is_cancelled: 

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

1300 

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

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

1303 

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

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

1306 ).scalar_one_or_none(): 

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

1308 

1309 session.add( 

1310 EventOrganizer( 

1311 user_id=request.user_id, 

1312 event_id=event.id, 

1313 ) 

1314 ) 

1315 session.flush() 

1316 

1317 other_user_context = make_notification_user_context(user_id=request.user_id) 

1318 

1319 notify( 

1320 session, 

1321 user_id=request.user_id, 

1322 topic_action=NotificationTopicAction.event__invite_organizer, 

1323 key=str(event.id), 

1324 data=notification_data_pb2.EventInviteOrganizer( 

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

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

1327 ), 

1328 ) 

1329 

1330 return empty_pb2.Empty() 

1331 

1332 def RemoveEventOrganizer( 

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

1334 ) -> empty_pb2.Empty: 

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

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

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

1338 

1339 event, occurrence = res 

1340 

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

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

1343 

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

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

1346 

1347 # Determine which user to remove 

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

1349 

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

1351 if event.owner_user_id == user_id_to_remove: 

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

1353 

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

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

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

1357 

1358 # Find the organizer to remove 

1359 organizer_to_remove = session.execute( 

1360 select(EventOrganizer) 

1361 .where(EventOrganizer.user_id == user_id_to_remove) 

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

1363 ).scalar_one_or_none() 

1364 

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

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

1367 

1368 session.delete(organizer_to_remove) 

1369 

1370 return empty_pb2.Empty() 

1371 

1372 def GetEventCalendarFile( 

1373 self, request: events_pb2.GetEventCalendarFileReq, context: CouchersContext, session: Session 

1374 ) -> httpbody_pb2.HttpBody: 

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

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

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

1378 

1379 _, occurrence_db = res 

1380 

1381 event_pb = event_to_pb(session, occurrence_db, context) 

1382 ics_data = create_event_ics_calendar(event_pb, context.localization).serialize().encode("utf-8") 

1383 return httpbody_pb2.HttpBody(content_type="text/calendar", data=ics_data)