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

558 statements  

« prev     ^ index     » next       coverage.py v7.15.2, created at 2026-07-25 18:54 +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.internal import jobs_pb2 

43from couchers.servicers.api import user_model_to_pb 

44from couchers.servicers.blocking import is_not_visible 

45from couchers.servicers.threads import thread_to_pb 

46from couchers.sql import users_visible, where_moderated_content_visible, where_users_column_visible 

47from couchers.tasks import send_event_community_invite_request_email 

48from couchers.utils import ( 

49 Timestamp_from_datetime, 

50 create_coordinate, 

51 datetime_to_iso8601_local, 

52 dt_from_millis, 

53 millis_from_dt, 

54 not_none, 

55 now, 

56) 

57 

58logger = logging.getLogger(__name__) 

59 

60attendancestate2sql = { 

61 events_pb2.AttendanceState.ATTENDANCE_STATE_NOT_GOING: None, 

62 events_pb2.AttendanceState.ATTENDANCE_STATE_GOING: AttendeeStatus.going, 

63} 

64 

65attendancestate2api = { 

66 None: events_pb2.AttendanceState.ATTENDANCE_STATE_NOT_GOING, 

67 AttendeeStatus.going: events_pb2.AttendanceState.ATTENDANCE_STATE_GOING, 

68} 

69 

70MAX_PAGINATION_LENGTH = 25 

71 

72 

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

74 """ 

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

76 """ 

77 if event.owner_user: 

78 return event.owner_user_id == user_id 

79 # otherwise owned by a cluster 

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

81 

82 

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

84 """ 

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

86 """ 

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

88 

89 

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

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

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

93 return True 

94 

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

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

97 

98 

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

100 return ( 

101 _is_event_owner(event, user_id) 

102 or _is_event_organizer(event, user_id) 

103 or _can_moderate_event(session, event, user_id) 

104 ) 

105 

106 

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

108 event = occurrence.event 

109 

110 next_occurrence = ( 

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

112 .order_by(EventOccurrence.end_time.asc()) 

113 .limit(1) 

114 .one_or_none() 

115 ) 

116 

117 owner_community_id = None 

118 owner_group_id = None 

119 if event.owner_cluster: 

120 if event.owner_cluster.is_official_cluster: 

121 owner_community_id = event.owner_cluster.parent_node_id 

122 else: 

123 owner_group_id = event.owner_cluster.id 

124 

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

126 attendance_state = attendance.attendee_status if attendance else None 

127 

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

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

130 

131 going_count = session.execute( 

132 where_users_column_visible( 

133 select(func.count()) 

134 .select_from(EventOccurrenceAttendee) 

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

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

137 context, 

138 EventOccurrenceAttendee.user_id, 

139 ) 

140 ).scalar_one() 

141 organizer_count = session.execute( 

142 where_users_column_visible( 

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

144 context, 

145 EventOrganizer.user_id, 

146 ) 

147 ).scalar_one() 

148 subscriber_count = session.execute( 

149 where_users_column_visible( 

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

151 context, 

152 EventSubscription.user_id, 

153 ) 

154 ).scalar_one() 

155 

156 return events_pb2.Event( 

157 event_id=occurrence.id, 

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

159 is_cancelled=occurrence.is_cancelled, 

160 is_deleted=occurrence.is_deleted, 

161 title=event.title, 

162 slug=event.slug, 

163 content=occurrence.content, 

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

165 photo_key=occurrence.photo_key or "", 

166 location=events_pb2.EventLocation( 

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

168 ), 

169 created=Timestamp_from_datetime(occurrence.created), 

170 last_edited=Timestamp_from_datetime(occurrence.last_edited), 

171 creator_user_id=occurrence.creator_user_id, 

172 start_time=Timestamp_from_datetime(occurrence.start_time), 

173 end_time=Timestamp_from_datetime(occurrence.end_time), 

174 timezone=occurrence.timezone, 

175 attendance_state=attendancestate2api[attendance_state], 

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

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

178 going_count=going_count, 

179 organizer_count=organizer_count, 

180 subscriber_count=subscriber_count, 

181 owner_user_id=event.owner_user_id, 

182 owner_community_id=owner_community_id, 

183 owner_group_id=owner_group_id, 

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

185 can_edit=can_edit, 

186 can_moderate=can_moderate, 

187 ) 

188 

189 

190def _get_event_and_occurrence_query( 

191 occurrence_id: int, 

192 include_deleted: bool, 

193 context: CouchersContext | None = None, 

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

195 query = ( 

196 select(Event, EventOccurrence) 

197 .where(EventOccurrence.id == occurrence_id) 

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

199 ) 

200 

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

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

203 

204 if context is not None: 

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

206 

207 return query 

208 

209 

210def _get_event_and_occurrence_one( 

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

212) -> tuple[Event, EventOccurrence]: 

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

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

215 return result._tuple() 

216 

217 

218def _get_event_and_occurrence_one_or_none( 

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

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

221 result = session.execute( 

222 _get_event_and_occurrence_query(occurrence_id, include_deleted, context=context) 

223 ).one_or_none() 

224 return result._tuple() if result else None 

225 

226 

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

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

229 if not location or not location.address: 

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

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

232 # No events allowed on Null Island 

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

234 

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

236 return (geom, location.address) 

237 

238 

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

240 timezone_id = session.execute( 

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

242 ).scalar_one_or_none() 

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

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

245 

246 return ZoneInfo(timezone_id) 

247 

248 

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

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

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

252 

253 try: 

254 naive_datetime = datetime.fromisoformat(value) 

255 except ValueError: 

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

257 

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

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

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

261 

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

263 

264 

265def _update_datetime( 

266 new_iso8601_local: str | None, 

267 new_timezone: ZoneInfo, 

268 old_datetime: datetime, 

269 old_timezone: ZoneInfo, 

270 context: CouchersContext, 

271) -> datetime: 

272 if new_iso8601_local is None and new_timezone != old_timezone: 

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

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

275 if new_iso8601_local is None: 

276 return old_datetime # No change 

277 # New effective datetime/timestamp 

278 return _check_iso8601_local_datetime(new_iso8601_local, new_timezone, context) 

279 

280 

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

282 if start_time < now(): 

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

284 if end_time < start_time: 

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

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

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

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

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

290 

291 

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

293 """ 

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

295 """ 

296 cluster = occurrence.event.parent_node.official_cluster 

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

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

299 return [], occurrence.event.parent_node_id 

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

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

302 else: 

303 max_radius = 20000 # m 

304 users = ( 

305 session.execute( 

306 select(User) 

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

308 .where(User.is_visible) 

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

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

311 ) 

312 .scalars() 

313 .all() 

314 ) 

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

316 

317 

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

319 """ 

320 Background job to generated/fan out event notifications 

321 """ 

322 # Import here to avoid circular dependency 

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

324 

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

326 

327 with session_scope() as session: 

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

329 creator = occurrence.creator_user 

330 

331 users, node_id = get_users_to_notify_for_new_event(session, occurrence) 

332 

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

334 

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

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

337 return 

338 

339 for user in users: 

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

341 continue 

342 context = make_notification_user_context(user_id=user.id) 

343 topic_action = ( 

344 NotificationTopicAction.event__create_approved 

345 if payload.approved 

346 else NotificationTopicAction.event__create_any 

347 ) 

348 notify( 

349 session, 

350 user_id=user.id, 

351 topic_action=topic_action, 

352 key=str(payload.occurrence_id), 

353 data=notification_data_pb2.EventCreate( 

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

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

356 nearby=True if node_id is None else None, 

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

358 ), 

359 moderation_state_id=occurrence.moderation_state_id, 

360 ) 

361 

362 

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

364 with session_scope() as session: 

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

366 

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

368 

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

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

371 

372 for user_id in set(subscribed_user_ids + attending_user_ids): 

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

374 continue 

375 context = make_notification_user_context(user_id=user_id) 

376 notify( 

377 session, 

378 user_id=user_id, 

379 topic_action=NotificationTopicAction.event__update, 

380 key=str(payload.occurrence_id), 

381 data=notification_data_pb2.EventUpdate( 

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

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

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

385 updated_str_items=payload.updated_str_items, 

386 updated_enum_items=( 

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

388 ), 

389 ), 

390 moderation_state_id=occurrence.moderation_state_id, 

391 ) 

392 

393 

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

395 with session_scope() as session: 

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

397 

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

399 

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

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

402 

403 for user_id in set(subscribed_user_ids + attending_user_ids): 

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

405 continue 

406 context = make_notification_user_context(user_id=user_id) 

407 notify( 

408 session, 

409 user_id=user_id, 

410 topic_action=NotificationTopicAction.event__cancel, 

411 key=str(payload.occurrence_id), 

412 data=notification_data_pb2.EventCancel( 

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

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

415 ), 

416 moderation_state_id=occurrence.moderation_state_id, 

417 ) 

418 

419 

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

421 with session_scope() as session: 

422 event, occurrence = _get_event_and_occurrence_one( 

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

424 ) 

425 

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

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

428 

429 for user_id in set(subscribed_user_ids + attending_user_ids): 

430 context = make_notification_user_context(user_id=user_id) 

431 notify( 

432 session, 

433 user_id=user_id, 

434 topic_action=NotificationTopicAction.event__delete, 

435 key=str(payload.occurrence_id), 

436 data=notification_data_pb2.EventDelete( 

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

438 ), 

439 moderation_state_id=occurrence.moderation_state_id, 

440 ) 

441 

442 

443class Events(events_pb2_grpc.EventsServicer): 

444 def CreateEvent( 

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

446 ) -> events_pb2.Event: 

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

448 if not has_completed_profile(session, user): 

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

450 if not request.title: 

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

452 if not request.content: 

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

454 

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

456 timezone = _check_timezone_at(geom, context, session) 

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

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

459 _check_occurrence_time_validity(start_datetime, end_datetime, context) 

460 

461 if request.parent_community_id: 

462 parent_node = session.execute( 

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

464 ).scalar_one_or_none() 

465 

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

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

468 else: 

469 # parent community computed from geom 

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

471 

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

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

474 

475 if ( 

476 request.photo_key 

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

478 ): 

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

480 

481 thread = Thread() 

482 session.add(thread) 

483 session.flush() 

484 

485 event = Event( 

486 title=request.title, 

487 parent_node_id=parent_node.id, 

488 owner_user_id=context.user_id, 

489 thread_id=thread.id, 

490 creator_user_id=context.user_id, 

491 ) 

492 session.add(event) 

493 session.flush() 

494 

495 occurrence: EventOccurrence | None = None 

496 

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

498 nonlocal occurrence 

499 occurrence = EventOccurrence( 

500 event_id=event.id, 

501 content=request.content, 

502 geom=geom, 

503 address=address, 

504 timezone=timezone.key, 

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

506 during=TimestamptzRange(start_datetime, end_datetime), 

507 creator_user_id=context.user_id, 

508 moderation_state_id=moderation_state_id, 

509 ) 

510 session.add(occurrence) 

511 session.flush() 

512 return occurrence.id 

513 

514 create_moderation( 

515 session=session, 

516 object_type=ModerationObjectType.event_occurrence, 

517 object_id=create_occurrence, 

518 creator_user_id=context.user_id, 

519 ) 

520 

521 assert occurrence is not None 

522 

523 session.add( 

524 EventOrganizer( 

525 user_id=context.user_id, 

526 event_id=event.id, 

527 ) 

528 ) 

529 

530 session.add( 

531 EventSubscription( 

532 user_id=context.user_id, 

533 event_id=event.id, 

534 ) 

535 ) 

536 

537 session.add( 

538 EventOccurrenceAttendee( 

539 user_id=context.user_id, 

540 occurrence_id=occurrence.id, 

541 attendee_status=AttendeeStatus.going, 

542 ) 

543 ) 

544 

545 session.commit() 

546 

547 log_event( 

548 context, 

549 session, 

550 "event.created", 

551 { 

552 "event_id": event.id, 

553 "occurrence_id": occurrence.id, 

554 "parent_community_id": parent_node.id, 

555 "parent_community_name": parent_node.official_cluster.name, 

556 }, 

557 ) 

558 

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

560 queue_job( 

561 session, 

562 job=generate_event_create_notifications, 

563 payload=jobs_pb2.GenerateEventCreateNotificationsPayload( 

564 inviting_user_id=user.id, 

565 occurrence_id=occurrence.id, 

566 approved=False, 

567 ), 

568 ) 

569 

570 return event_to_pb(session, occurrence, context) 

571 

572 def ScheduleEvent( 

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

574 ) -> events_pb2.Event: 

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

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

577 

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

579 timezone = _check_timezone_at(geom, context, session) 

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

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

582 _check_occurrence_time_validity(start_datetime, end_datetime, context) 

583 

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

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

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

587 

588 event, occurrence = res 

589 

590 if not _can_edit_event(session, event, context.user_id): 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.PERMISSION_DENIED, "event_edit_permission_denied") 

592 

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

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

595 

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

597 request.photo_key 

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

599 ): 

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

601 

602 during = TimestamptzRange(start_datetime, end_datetime) 

603 

604 # && is the overlap operator for ranges 

605 if ( 

606 session.execute( 

607 select(EventOccurrence.id) 

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

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

610 .limit(1) 

611 ) 

612 .scalars() 

613 .one_or_none() 

614 is not None 

615 ): 

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

617 

618 new_occurrence: EventOccurrence | None = None 

619 

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

621 nonlocal new_occurrence 

622 new_occurrence = EventOccurrence( 

623 event_id=event.id, 

624 content=request.content, 

625 geom=geom, 

626 address=address, 

627 timezone=timezone.key, 

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

629 during=during, 

630 creator_user_id=context.user_id, 

631 moderation_state_id=moderation_state_id, 

632 ) 

633 session.add(new_occurrence) 

634 session.flush() 

635 return new_occurrence.id 

636 

637 create_moderation( 

638 session=session, 

639 object_type=ModerationObjectType.event_occurrence, 

640 object_id=create_occurrence, 

641 creator_user_id=context.user_id, 

642 ) 

643 

644 assert new_occurrence is not None 

645 

646 session.add( 

647 EventOccurrenceAttendee( 

648 user_id=context.user_id, 

649 occurrence_id=new_occurrence.id, 

650 attendee_status=AttendeeStatus.going, 

651 ) 

652 ) 

653 

654 session.flush() 

655 

656 # TODO: notify 

657 

658 return event_to_pb(session, new_occurrence, context) 

659 

660 def UpdateEvent( 

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

662 ) -> events_pb2.Event: 

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

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

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

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

667 

668 event, occurrence = res 

669 

670 if not _can_edit_event(session, event, context.user_id): 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.PERMISSION_DENIED, "event_edit_permission_denied") 

672 

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

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

675 

676 if occurrence.is_cancelled: 

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

678 

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

680 

681 if request.HasField("title"): 

682 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_TITLE) 

683 event.title = request.title.value 

684 

685 if request.HasField("content"): 

686 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_CONTENT) 

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

688 

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

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

691 

692 old_timezone = ZoneInfo(occurrence.timezone) 

693 timezone: ZoneInfo = old_timezone 

694 if request.HasField("location"): 

695 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_LOCATION) 

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

697 timezone = _check_timezone_at(geom, context, session) 

698 occurrence_update["geom"] = geom 

699 occurrence_update["address"] = address 

700 occurrence_update["timezone"] = timezone.key 

701 

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

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

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

705 

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

707 start_datetime = _update_datetime( 

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

709 timezone, 

710 old_datetime=occurrence.start_time, 

711 old_timezone=old_timezone, 

712 context=context, 

713 ) 

714 if start_datetime != occurrence.start_time: 

715 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_START_TIME) 

716 

717 end_datetime = _update_datetime( 

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

719 timezone, 

720 old_datetime=occurrence.end_time, 

721 old_timezone=old_timezone, 

722 context=context, 

723 ) 

724 if end_datetime != occurrence.end_time: 

725 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_END_TIME) 

726 

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

728 _check_occurrence_time_validity(start_datetime, end_datetime, context) 

729 

730 during = TimestamptzRange(start_datetime, end_datetime) 

731 

732 # && is the overlap operator for ranges 

733 if ( 

734 session.execute( 

735 select(EventOccurrence.id) 

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

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

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

739 .limit(1) 

740 ) 

741 .scalars() 

742 .one_or_none() 

743 is not None 

744 ): 

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

746 

747 occurrence_update["during"] = during 

748 

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

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

751 

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

753 if request.update_all_future: 

754 session.execute( 

755 update(EventOccurrence) 

756 .where(EventOccurrence.end_time >= cutoff_time) 

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

758 .values(occurrence_update) 

759 .execution_options(synchronize_session=False) 

760 ) 

761 else: 

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

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

764 session.execute( 

765 update(EventOccurrence) 

766 .where(EventOccurrence.end_time >= cutoff_time) 

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

768 .values(occurrence_update) 

769 .execution_options(synchronize_session=False) 

770 ) 

771 

772 session.flush() 

773 

774 if notify_updated: 

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

776 if request.should_notify: 

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

778 

779 queue_job( 

780 session, 

781 job=generate_event_update_notifications, 

782 payload=jobs_pb2.GenerateEventUpdateNotificationsPayload( 

783 updating_user_id=user.id, 

784 occurrence_id=occurrence.id, 

785 updated_enum_items=notify_updated, 

786 ), 

787 ) 

788 else: 

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

790 

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

792 session.refresh(occurrence) 

793 

794 return event_to_pb(session, occurrence, context) 

795 

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

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

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

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

800 

801 if not occurrence: 

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

803 

804 return event_to_pb(session, occurrence, context) 

805 

806 def CancelEvent( 

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

808 ) -> empty_pb2.Empty: 

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

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

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

812 

813 event, occurrence = res 

814 

815 if not _can_edit_event(session, event, context.user_id): 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.PERMISSION_DENIED, "event_edit_permission_denied") 

817 

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

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

820 

821 occurrence.is_cancelled = True 

822 

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

824 

825 queue_job( 

826 session, 

827 job=generate_event_cancel_notifications, 

828 payload=jobs_pb2.GenerateEventCancelNotificationsPayload( 

829 cancelling_user_id=context.user_id, 

830 occurrence_id=occurrence.id, 

831 ), 

832 ) 

833 

834 return empty_pb2.Empty() 

835 

836 def RequestCommunityInvite( 

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

838 ) -> empty_pb2.Empty: 

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

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

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

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

843 

844 event, occurrence = res 

845 

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

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

848 

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

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

851 

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

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

854 

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

856 

857 if len(this_user_reqs) > 0: 

858 context.abort_with_error_code( 

859 grpc.StatusCode.FAILED_PRECONDITION, "event_community_invite_already_requested" 

860 ) 

861 

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

863 

864 if len(approved_reqs) > 0: 

865 context.abort_with_error_code( 

866 grpc.StatusCode.FAILED_PRECONDITION, "event_community_invite_already_approved" 

867 ) 

868 

869 req = EventCommunityInviteRequest( 

870 occurrence_id=request.event_id, 

871 user_id=context.user_id, 

872 ) 

873 session.add(req) 

874 session.flush() 

875 

876 send_event_community_invite_request_email(session, req) 

877 

878 return empty_pb2.Empty() 

879 

880 def ListEventOccurrences( 

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

882 ) -> events_pb2.ListEventOccurrencesRes: 

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

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

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

886 initial_query = ( 

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

888 ) 

889 initial_query = where_moderated_content_visible( 

890 initial_query, context, EventOccurrence, is_list_operation=False 

891 ) 

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

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

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

895 

896 query = ( 

897 select(EventOccurrence) 

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

899 .where(~EventOccurrence.is_deleted) 

900 ) 

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

902 

903 if not request.include_cancelled: 

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

905 

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

907 cutoff = page_token - timedelta(seconds=1) 

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

909 else: 

910 cutoff = page_token + timedelta(seconds=1) 

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

912 

913 query = query.limit(page_size + 1) 

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

915 

916 return events_pb2.ListEventOccurrencesRes( 

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

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

919 ) 

920 

921 def ListEventAttendees( 

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

923 ) -> events_pb2.ListEventAttendeesRes: 

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

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

926 occurrence = session.execute( 

927 where_moderated_content_visible( 

928 select(EventOccurrence) 

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

930 .where(~EventOccurrence.is_deleted), 

931 context, 

932 EventOccurrence, 

933 is_list_operation=False, 

934 ) 

935 ).scalar_one_or_none() 

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

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

938 attendees = ( 

939 session.execute( 

940 where_users_column_visible( 

941 select(EventOccurrenceAttendee) 

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

943 .where(EventOccurrenceAttendee.user_id >= next_user_id) 

944 .order_by(EventOccurrenceAttendee.user_id) 

945 .limit(page_size + 1), 

946 context, 

947 EventOccurrenceAttendee.user_id, 

948 ) 

949 ) 

950 .scalars() 

951 .all() 

952 ) 

953 return events_pb2.ListEventAttendeesRes( 

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

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

956 ) 

957 

958 def ListEventSubscribers( 

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

960 ) -> events_pb2.ListEventSubscribersRes: 

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

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

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

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

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

966 event, occurrence = res 

967 subscribers = ( 

968 session.execute( 

969 where_users_column_visible( 

970 select(EventSubscription) 

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

972 .where(EventSubscription.user_id >= next_user_id) 

973 .order_by(EventSubscription.user_id) 

974 .limit(page_size + 1), 

975 context, 

976 EventSubscription.user_id, 

977 ) 

978 ) 

979 .scalars() 

980 .all() 

981 ) 

982 return events_pb2.ListEventSubscribersRes( 

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

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

985 ) 

986 

987 def ListEventOrganizers( 

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

989 ) -> events_pb2.ListEventOrganizersRes: 

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

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

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

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

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

995 event, occurrence = res 

996 organizers = ( 

997 session.execute( 

998 where_users_column_visible( 

999 select(EventOrganizer) 

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

1001 .where(EventOrganizer.user_id >= next_user_id) 

1002 .order_by(EventOrganizer.user_id) 

1003 .limit(page_size + 1), 

1004 context, 

1005 EventOrganizer.user_id, 

1006 ) 

1007 ) 

1008 .scalars() 

1009 .all() 

1010 ) 

1011 return events_pb2.ListEventOrganizersRes( 

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

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

1014 ) 

1015 

1016 def TransferEvent( 

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

1018 ) -> events_pb2.Event: 

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

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

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

1022 

1023 event, occurrence = res 

1024 

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

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

1027 

1028 if occurrence.is_cancelled: 

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

1030 

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

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

1033 

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

1035 cluster = session.execute( 

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

1037 ).scalar_one_or_none() 

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

1039 cluster = session.execute( 

1040 select(Cluster) 

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

1042 .where(Cluster.is_official_cluster) 

1043 ).scalar_one_or_none() 

1044 

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

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

1047 

1048 event.owner_user = None 

1049 event.owner_cluster = cluster 

1050 

1051 session.commit() 

1052 return event_to_pb(session, occurrence, context) 

1053 

1054 def SetEventSubscription( 

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

1056 ) -> events_pb2.Event: 

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

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

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

1060 

1061 event, occurrence = res 

1062 

1063 if occurrence.is_cancelled: 

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

1065 

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

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

1068 

1069 current_subscription = session.execute( 

1070 select(EventSubscription) 

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

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

1073 ).scalar_one_or_none() 

1074 

1075 # if not subscribed, subscribe 

1076 if request.subscribe and not current_subscription: 

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

1078 

1079 # if subscribed but unsubbing, remove subscription 

1080 if not request.subscribe and current_subscription: 

1081 session.delete(current_subscription) 

1082 

1083 session.flush() 

1084 

1085 log_event( 

1086 context, 

1087 session, 

1088 "event.subscription_set", 

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

1090 ) 

1091 

1092 return event_to_pb(session, occurrence, context) 

1093 

1094 def SetEventAttendance( 

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

1096 ) -> events_pb2.Event: 

1097 occurrence = session.execute( 

1098 where_moderated_content_visible( 

1099 select(EventOccurrence) 

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

1101 .where(~EventOccurrence.is_deleted), 

1102 context, 

1103 EventOccurrence, 

1104 is_list_operation=False, 

1105 ) 

1106 ).scalar_one_or_none() 

1107 

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

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

1110 

1111 if occurrence.is_cancelled: 

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

1113 

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

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

1116 

1117 current_attendance = session.execute( 

1118 select(EventOccurrenceAttendee) 

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

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

1121 ).scalar_one_or_none() 

1122 

1123 if request.attendance_state == events_pb2.ATTENDANCE_STATE_NOT_GOING: 

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

1125 session.delete(current_attendance) 

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

1127 else: 

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

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

1130 else: 

1131 # create new 

1132 attendance = EventOccurrenceAttendee( 

1133 user_id=context.user_id, 

1134 occurrence_id=occurrence.id, 

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

1136 ) 

1137 session.add(attendance) 

1138 

1139 session.flush() 

1140 

1141 log_event( 

1142 context, 

1143 session, 

1144 "event.attendance_set", 

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

1146 ) 

1147 

1148 return event_to_pb(session, occurrence, context) 

1149 

1150 def ListMyEvents( 

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

1152 ) -> events_pb2.ListMyEventsRes: 

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

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

1155 page_token = ( 

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

1157 ) 

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

1159 page_number = request.page_number or 1 

1160 # Calculate the offset for pagination 

1161 offset = (page_number - 1) * page_size 

1162 query = ( 

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

1164 ) 

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

1166 

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

1168 include_subscribed = request.subscribed or include_all 

1169 include_organizing = request.organizing or include_all 

1170 include_attending = request.attending or include_all 

1171 include_my_communities = request.my_communities or include_all 

1172 

1173 if include_attending and request.exclude_attending: 

1174 context.abort_with_error_code( 

1175 grpc.StatusCode.INVALID_ARGUMENT, "cannot_combine_attending_and_exclude_attending" 

1176 ) 

1177 

1178 where_ = [] 

1179 

1180 if include_subscribed: 

1181 query = query.outerjoin( 

1182 EventSubscription, 

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

1184 ) 

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

1186 if include_organizing: 

1187 query = query.outerjoin( 

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

1189 ) 

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

1191 if include_attending or request.exclude_attending: 

1192 query = query.outerjoin( 

1193 EventOccurrenceAttendee, 

1194 and_( 

1195 EventOccurrenceAttendee.occurrence_id == EventOccurrence.id, 

1196 EventOccurrenceAttendee.user_id == context.user_id, 

1197 ), 

1198 ) 

1199 if include_attending: 

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

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

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

1203 query = query.outerjoin( 

1204 EventOrganizer, 

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

1206 ) 

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

1208 if include_my_communities: 

1209 my_communities = ( 

1210 session.execute( 

1211 select(Node.id) 

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

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

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

1215 .where(Cluster.is_official_cluster) 

1216 .order_by(Node.id) 

1217 .limit(100000) 

1218 ) 

1219 .scalars() 

1220 .all() 

1221 ) 

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

1223 

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

1225 

1226 if request.my_communities_exclude_global: 

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

1228 

1229 if not request.include_cancelled: 

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

1231 

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

1233 cutoff = page_token - timedelta(seconds=1) 

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

1235 else: 

1236 cutoff = page_token + timedelta(seconds=1) 

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

1238 # Count the total number of items for pagination 

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

1240 # Apply pagination by page number 

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

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

1243 

1244 return events_pb2.ListMyEventsRes( 

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

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

1247 total_items=total_items, 

1248 ) 

1249 

1250 def ListAllEvents( 

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

1252 ) -> events_pb2.ListAllEventsRes: 

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

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

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

1256 

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

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

1259 

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

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

1262 

1263 if not request.past: 

1264 cutoff = page_token - timedelta(seconds=1) 

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

1266 else: 

1267 cutoff = page_token + timedelta(seconds=1) 

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

1269 

1270 query = query.limit(page_size + 1) 

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

1272 

1273 return events_pb2.ListAllEventsRes( 

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

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

1276 ) 

1277 

1278 def InviteEventOrganizer( 

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

1280 ) -> empty_pb2.Empty: 

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

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

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

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

1285 

1286 event, occurrence = res 

1287 

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

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

1290 

1291 if occurrence.is_cancelled: 

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

1293 

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

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

1296 

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

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

1299 ).scalar_one_or_none(): 

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

1301 

1302 session.add( 

1303 EventOrganizer( 

1304 user_id=request.user_id, 

1305 event_id=event.id, 

1306 ) 

1307 ) 

1308 session.flush() 

1309 

1310 other_user_context = make_notification_user_context(user_id=request.user_id) 

1311 

1312 notify( 

1313 session, 

1314 user_id=request.user_id, 

1315 topic_action=NotificationTopicAction.event__invite_organizer, 

1316 key=str(event.id), 

1317 data=notification_data_pb2.EventInviteOrganizer( 

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

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

1320 ), 

1321 ) 

1322 

1323 return empty_pb2.Empty() 

1324 

1325 def RemoveEventOrganizer( 

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

1327 ) -> empty_pb2.Empty: 

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

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

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

1331 

1332 event, occurrence = res 

1333 

1334 if occurrence.is_cancelled: 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.PERMISSION_DENIED, "event_cant_update_cancelled_event") 

1336 

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

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

1339 

1340 # Determine which user to remove 

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

1342 

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

1344 if event.owner_user_id == user_id_to_remove: 

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

1346 

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

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

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

1350 

1351 # Find the organizer to remove 

1352 organizer_to_remove = session.execute( 

1353 select(EventOrganizer) 

1354 .where(EventOrganizer.user_id == user_id_to_remove) 

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

1356 ).scalar_one_or_none() 

1357 

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

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

1360 

1361 session.delete(organizer_to_remove) 

1362 

1363 return empty_pb2.Empty() 

1364 

1365 def GetEventCalendarFile( 

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

1367 ) -> events_pb2.GetEventCalendarFileRes: 

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

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

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

1371 

1372 _, occurrence_db = res 

1373 

1374 event_pb = event_to_pb(session, occurrence_db, context) 

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

1376 return events_pb2.GetEventCalendarFileRes(data=ics_data)