Coverage for app/backend/src/couchers/servicers/events.py: 85%
550 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-25 22:54 +0000
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-25 22:54 +0000
1import logging
2from datetime import datetime, timedelta
3from typing import Any, cast
4from zoneinfo import ZoneInfo
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
14from couchers.context import CouchersContext, make_notification_user_context
15from couchers.db import can_moderate_node, get_parent_node_at_location, session_scope
16from couchers.event_log import log_event
17from couchers.helpers.completed_profile import has_completed_profile
18from couchers.jobs.enqueue import queue_job
19from couchers.models import (
20 AttendeeStatus,
21 Cluster,
22 ClusterSubscription,
23 Event,
24 EventCommunityInviteRequest,
25 EventOccurrence,
26 EventOccurrenceAttendee,
27 EventOrganizer,
28 EventSubscription,
29 ModerationObjectType,
30 Node,
31 NodeType,
32 Thread,
33 Upload,
34 User,
35)
36from couchers.models.notifications import NotificationTopicAction
37from couchers.models.static import TimezoneArea
38from couchers.moderation.utils import create_moderation
39from couchers.notifications.notify import notify
40from couchers.proto import events_pb2, events_pb2_grpc, notification_data_pb2
41from couchers.proto.internal import jobs_pb2
42from couchers.servicers.api import user_model_to_pb
43from couchers.servicers.blocking import is_not_visible
44from couchers.servicers.threads import thread_to_pb
45from couchers.sql import users_visible, where_moderated_content_visible, where_users_column_visible
46from couchers.tasks import send_event_community_invite_request_email
47from couchers.utils import (
48 Timestamp_from_datetime,
49 create_coordinate,
50 datetime_to_iso8601_local,
51 dt_from_millis,
52 millis_from_dt,
53 not_none,
54 now,
55)
57logger = logging.getLogger(__name__)
59attendancestate2sql = {
60 events_pb2.AttendanceState.ATTENDANCE_STATE_NOT_GOING: None,
61 events_pb2.AttendanceState.ATTENDANCE_STATE_GOING: AttendeeStatus.going,
62}
64attendancestate2api = {
65 None: events_pb2.AttendanceState.ATTENDANCE_STATE_NOT_GOING,
66 AttendeeStatus.going: events_pb2.AttendanceState.ATTENDANCE_STATE_GOING,
67}
69MAX_PAGINATION_LENGTH = 25
72def _is_event_owner(event: Event, user_id: int) -> bool:
73 """
74 Checks whether the user can act as an owner of the event
75 """
76 if event.owner_user:
77 return event.owner_user_id == user_id
78 # otherwise owned by a cluster
79 return not_none(event.owner_cluster).admins.where(User.id == user_id).one_or_none() is not None
82def _is_event_organizer(event: Event, user_id: int) -> bool:
83 """
84 Checks whether the user is as an organizer of the event
85 """
86 return event.organizers.where(EventOrganizer.user_id == user_id).one_or_none() is not None
89def _can_moderate_event(session: Session, event: Event, user_id: int) -> bool:
90 # if the event is owned by a cluster, then any moderator of that cluster can moderate this event
91 if event.owner_cluster is not None and can_moderate_node(session, user_id, event.owner_cluster.parent_node_id):
92 return True
94 # finally check if the user can moderate the parent node of the cluster
95 return can_moderate_node(session, user_id, event.parent_node_id)
98def _can_edit_event(session: Session, event: Event, user_id: int) -> bool:
99 return (
100 _is_event_owner(event, user_id)
101 or _is_event_organizer(event, user_id)
102 or _can_moderate_event(session, event, user_id)
103 )
106def event_to_pb(session: Session, occurrence: EventOccurrence, context: CouchersContext) -> events_pb2.Event:
107 event = occurrence.event
109 next_occurrence = (
110 event.occurrences.where(EventOccurrence.end_time >= now())
111 .order_by(EventOccurrence.end_time.asc())
112 .limit(1)
113 .one_or_none()
114 )
116 owner_community_id = None
117 owner_group_id = None
118 if event.owner_cluster:
119 if event.owner_cluster.is_official_cluster:
120 owner_community_id = event.owner_cluster.parent_node_id
121 else:
122 owner_group_id = event.owner_cluster.id
124 attendance = occurrence.attendances.where(EventOccurrenceAttendee.user_id == context.user_id).one_or_none()
125 attendance_state = attendance.attendee_status if attendance else None
127 can_moderate = _can_moderate_event(session, event, context.user_id)
128 can_edit = _can_edit_event(session, event, context.user_id)
130 going_count = session.execute(
131 where_users_column_visible(
132 select(func.count())
133 .select_from(EventOccurrenceAttendee)
134 .where(EventOccurrenceAttendee.occurrence_id == occurrence.id)
135 .where(EventOccurrenceAttendee.attendee_status == AttendeeStatus.going),
136 context,
137 EventOccurrenceAttendee.user_id,
138 )
139 ).scalar_one()
140 organizer_count = session.execute(
141 where_users_column_visible(
142 select(func.count()).select_from(EventOrganizer).where(EventOrganizer.event_id == event.id),
143 context,
144 EventOrganizer.user_id,
145 )
146 ).scalar_one()
147 subscriber_count = session.execute(
148 where_users_column_visible(
149 select(func.count()).select_from(EventSubscription).where(EventSubscription.event_id == event.id),
150 context,
151 EventSubscription.user_id,
152 )
153 ).scalar_one()
155 return events_pb2.Event(
156 event_id=occurrence.id,
157 is_next=False if not next_occurrence else occurrence.id == next_occurrence.id,
158 is_cancelled=occurrence.is_cancelled,
159 is_deleted=occurrence.is_deleted,
160 title=event.title,
161 slug=event.slug,
162 content=occurrence.content,
163 photo_url=occurrence.photo.full_url if occurrence.photo else None,
164 photo_key=occurrence.photo_key or "",
165 location=events_pb2.EventLocation(
166 lat=occurrence.coordinates[0], lng=occurrence.coordinates[1], address=occurrence.address
167 ),
168 created=Timestamp_from_datetime(occurrence.created),
169 last_edited=Timestamp_from_datetime(occurrence.last_edited),
170 creator_user_id=occurrence.creator_user_id,
171 start_time=Timestamp_from_datetime(occurrence.start_time),
172 end_time=Timestamp_from_datetime(occurrence.end_time),
173 timezone=occurrence.timezone,
174 attendance_state=attendancestate2api[attendance_state],
175 organizer=event.organizers.where(EventOrganizer.user_id == context.user_id).one_or_none() is not None,
176 subscriber=event.subscribers.where(EventSubscription.user_id == context.user_id).one_or_none() is not None,
177 going_count=going_count,
178 organizer_count=organizer_count,
179 subscriber_count=subscriber_count,
180 owner_user_id=event.owner_user_id,
181 owner_community_id=owner_community_id,
182 owner_group_id=owner_group_id,
183 thread=thread_to_pb(session, context, event.thread_id),
184 can_edit=can_edit,
185 can_moderate=can_moderate,
186 )
189def _get_event_and_occurrence_query(
190 occurrence_id: int,
191 include_deleted: bool,
192 context: CouchersContext | None = None,
193) -> Select[tuple[Event, EventOccurrence]]:
194 query = (
195 select(Event, EventOccurrence)
196 .where(EventOccurrence.id == occurrence_id)
197 .where(EventOccurrence.event_id == Event.id)
198 )
200 if not include_deleted: 200 ↛ 203line 200 didn't jump to line 203 because the condition on line 200 was always true
201 query = query.where(~EventOccurrence.is_deleted)
203 if context is not None:
204 query = where_moderated_content_visible(query, context, EventOccurrence, is_list_operation=False)
206 return query
209def _get_event_and_occurrence_one(
210 session: Session, occurrence_id: int, include_deleted: bool = False
211) -> tuple[Event, EventOccurrence]:
212 """For background jobs only - no visibility filtering."""
213 result = session.execute(_get_event_and_occurrence_query(occurrence_id, include_deleted)).one()
214 return result._tuple()
217def _get_event_and_occurrence_one_or_none(
218 session: Session, occurrence_id: int, context: CouchersContext, include_deleted: bool = False
219) -> tuple[Event, EventOccurrence] | None:
220 result = session.execute(
221 _get_event_and_occurrence_query(occurrence_id, include_deleted, context=context)
222 ).one_or_none()
223 return result._tuple() if result else None
226def _check_location(location: events_pb2.EventLocation | None, context: CouchersContext) -> tuple[WKBElement, str]:
227 # As protobuf parses a missing value as 0.0, this is not a permitted event coordinate value
228 if not location or not location.address:
229 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "missing_event_address_or_location")
230 if location.lat == 0 and location.lng == 0:
231 # No events allowed on Null Island
232 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_coordinate")
234 geom = create_coordinate(location.lat, location.lng)
235 return (geom, location.address)
238def _check_timezone_at(geom: WKBElement, context: CouchersContext, session: Session) -> ZoneInfo:
239 timezone_id = session.execute(
240 select(TimezoneArea.tzid).where(func.ST_Contains(TimezoneArea.geom, func.ST_PointOnSurface(geom))).limit(1)
241 ).scalar_one_or_none()
242 if not timezone_id: 242 ↛ 243line 242 didn't jump to line 243 because the condition on line 242 was never true
243 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "event_timezone_not_found")
245 return ZoneInfo(timezone_id)
248def _check_iso8601_local_datetime(value: str, timezone: ZoneInfo, context: CouchersContext) -> datetime:
249 if not value: 249 ↛ 250line 249 didn't jump to line 250 because the condition on line 249 was never true
250 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "missing_event_start_end_datetime")
252 try:
253 naive_datetime = datetime.fromisoformat(value)
254 except ValueError:
255 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_event_start_end_datetime")
257 if naive_datetime.tzinfo is not None: 257 ↛ 259line 257 didn't jump to line 259 because the condition on line 257 was never true
258 # Expected a local datetime, otherwise we have two sources of timezones.
259 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_event_start_end_datetime")
261 return naive_datetime.replace(tzinfo=timezone).replace(second=0, microsecond=0)
264def _update_datetime(
265 new_iso8601_local: str | None,
266 new_timezone: ZoneInfo,
267 old_datetime: datetime,
268 old_timezone: ZoneInfo,
269 context: CouchersContext,
270) -> datetime:
271 if new_iso8601_local is None and new_timezone != old_timezone:
272 # Local time wasn't updated, but the timezone changed so the effective datetime/timestamp may have changed.
273 new_iso8601_local = datetime_to_iso8601_local(old_datetime.astimezone(old_timezone))
274 if new_iso8601_local is None:
275 return old_datetime # No change
276 # New effective datetime/timestamp
277 return _check_iso8601_local_datetime(new_iso8601_local, new_timezone, context)
280def _check_occurrence_time_validity(start_time: datetime, end_time: datetime, context: CouchersContext) -> None:
281 if start_time < now():
282 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "event_in_past")
283 if end_time < start_time:
284 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "event_ends_before_starts")
285 if end_time - start_time > timedelta(days=7):
286 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "event_too_long")
287 if start_time - now() > timedelta(days=365):
288 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "event_too_far_in_future")
291def get_users_to_notify_for_new_event(session: Session, occurrence: EventOccurrence) -> tuple[list[User], int | None]:
292 """
293 Returns the users to notify, as well as the community id that is being notified (None if based on geo search)
294 """
295 # people already attending or organizing the event don't need an invite to it
296 not_already_involved = User.id.not_in(
297 select(EventOccurrenceAttendee.user_id)
298 .where(EventOccurrenceAttendee.occurrence_id == occurrence.id)
299 .union(select(EventOrganizer.user_id).where(EventOrganizer.event_id == occurrence.event_id))
300 )
302 cluster = occurrence.event.parent_node.official_cluster
303 if occurrence.event.parent_node.node_type.value <= NodeType.region.value:
304 logger.info("Global, macroregion, and region communities are too big for email notifications.")
305 return [], occurrence.event.parent_node_id
306 elif occurrence.creator_user in cluster.admins or cluster.is_leaf: 306 ↛ 309line 306 didn't jump to line 309 because the condition on line 306 was always true
307 return list(cluster.members.where(User.is_visible).where(not_already_involved)), occurrence.event.parent_node_id
308 else:
309 max_radius = 20000 # m
310 users = (
311 session.execute(
312 select(User)
313 .join(ClusterSubscription, ClusterSubscription.user_id == User.id)
314 .where(User.is_visible)
315 .where(ClusterSubscription.cluster_id == cluster.id)
316 .where(func.ST_DWithin(User.geom, occurrence.geom, max_radius / 111111))
317 .where(not_already_involved)
318 )
319 .scalars()
320 .all()
321 )
322 return cast(tuple[list[User], int | None], (users, None))
325def generate_event_create_notifications(payload: jobs_pb2.GenerateEventCreateNotificationsPayload) -> None:
326 """
327 Background job to generated/fan out event notifications
328 """
329 # Import here to avoid circular dependency
330 from couchers.servicers.communities import community_to_pb # noqa: PLC0415
332 logger.info(f"Fanning out notifications for event occurrence id = {payload.occurrence_id}")
334 with session_scope() as session:
335 event, occurrence = _get_event_and_occurrence_one(session, occurrence_id=payload.occurrence_id)
336 creator = occurrence.creator_user
338 users, node_id = get_users_to_notify_for_new_event(session, occurrence)
340 inviting_user = session.execute(select(User).where(User.id == payload.inviting_user_id)).scalar_one_or_none()
342 if not inviting_user: 342 ↛ 343line 342 didn't jump to line 343 because the condition on line 342 was never true
343 logger.error(f"Inviting user {payload.inviting_user_id} is gone while trying to send event notification?")
344 return
346 for user in users:
347 if is_not_visible(session, user.id, creator.id): 347 ↛ 348line 347 didn't jump to line 348 because the condition on line 347 was never true
348 continue
349 context = make_notification_user_context(user_id=user.id)
350 topic_action = (
351 NotificationTopicAction.event__create_approved
352 if payload.approved
353 else NotificationTopicAction.event__create_any
354 )
355 notify(
356 session,
357 user_id=user.id,
358 topic_action=topic_action,
359 key=str(payload.occurrence_id),
360 data=notification_data_pb2.EventCreate(
361 event=event_to_pb(session, occurrence, context),
362 inviting_user=user_model_to_pb(inviting_user, session, context),
363 nearby=True if node_id is None else None,
364 in_community=community_to_pb(session, event.parent_node, context) if node_id is not None else None,
365 ),
366 moderation_state_id=occurrence.moderation_state_id,
367 )
370def generate_event_update_notifications(payload: jobs_pb2.GenerateEventUpdateNotificationsPayload) -> None:
371 with session_scope() as session:
372 event, occurrence = _get_event_and_occurrence_one(session, occurrence_id=payload.occurrence_id)
374 updating_user = session.execute(select(User).where(User.id == payload.updating_user_id)).scalar_one()
376 subscribed_user_ids = [user.id for user in event.subscribers]
377 attending_user_ids = [user.user_id for user in occurrence.attendances]
379 for user_id in set(subscribed_user_ids + attending_user_ids):
380 if is_not_visible(session, user_id, updating_user.id): 380 ↛ 381line 380 didn't jump to line 381 because the condition on line 380 was never true
381 continue
382 context = make_notification_user_context(user_id=user_id)
383 notify(
384 session,
385 user_id=user_id,
386 topic_action=NotificationTopicAction.event__update,
387 key=str(payload.occurrence_id),
388 data=notification_data_pb2.EventUpdate(
389 event=event_to_pb(session, occurrence, context),
390 updating_user=user_model_to_pb(updating_user, session, context),
391 updated_enum_items=(
392 notification_data_pb2.EventUpdateItem.ValueType(value) for value in payload.updated_enum_items
393 ),
394 ),
395 moderation_state_id=occurrence.moderation_state_id,
396 )
399def generate_event_cancel_notifications(payload: jobs_pb2.GenerateEventCancelNotificationsPayload) -> None:
400 with session_scope() as session:
401 event, occurrence = _get_event_and_occurrence_one(session, occurrence_id=payload.occurrence_id)
403 cancelling_user = session.execute(select(User).where(User.id == payload.cancelling_user_id)).scalar_one()
405 subscribed_user_ids = [user.id for user in event.subscribers]
406 attending_user_ids = [user.user_id for user in occurrence.attendances]
408 for user_id in set(subscribed_user_ids + attending_user_ids):
409 if is_not_visible(session, user_id, cancelling_user.id): 409 ↛ 410line 409 didn't jump to line 410 because the condition on line 409 was never true
410 continue
411 context = make_notification_user_context(user_id=user_id)
412 notify(
413 session,
414 user_id=user_id,
415 topic_action=NotificationTopicAction.event__cancel,
416 key=str(payload.occurrence_id),
417 data=notification_data_pb2.EventCancel(
418 event=event_to_pb(session, occurrence, context),
419 cancelling_user=user_model_to_pb(cancelling_user, session, context),
420 ),
421 moderation_state_id=occurrence.moderation_state_id,
422 )
425def generate_event_delete_notifications(payload: jobs_pb2.GenerateEventDeleteNotificationsPayload) -> None:
426 with session_scope() as session:
427 event, occurrence = _get_event_and_occurrence_one(
428 session, occurrence_id=payload.occurrence_id, include_deleted=True
429 )
431 subscribed_user_ids = [user.id for user in event.subscribers]
432 attending_user_ids = [user.user_id for user in occurrence.attendances]
434 for user_id in set(subscribed_user_ids + attending_user_ids):
435 context = make_notification_user_context(user_id=user_id)
436 notify(
437 session,
438 user_id=user_id,
439 topic_action=NotificationTopicAction.event__delete,
440 key=str(payload.occurrence_id),
441 data=notification_data_pb2.EventDelete(
442 event=event_to_pb(session, occurrence, context),
443 ),
444 moderation_state_id=occurrence.moderation_state_id,
445 )
448class Events(events_pb2_grpc.EventsServicer):
449 def CreateEvent(
450 self, request: events_pb2.CreateEventReq, context: CouchersContext, session: Session
451 ) -> events_pb2.Event:
452 user = session.execute(select(User).where(User.id == context.user_id)).scalar_one()
453 if not has_completed_profile(session, user):
454 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "incomplete_profile_create_event")
455 if not request.title:
456 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "missing_event_title")
457 if not request.content:
458 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "missing_event_content")
460 geom, address = _check_location(request.location if request.HasField("location") else None, context)
461 timezone = _check_timezone_at(geom, context, session)
462 start_datetime = _check_iso8601_local_datetime(request.start_datetime_iso8601_local, timezone, context)
463 end_datetime = _check_iso8601_local_datetime(request.end_datetime_iso8601_local, timezone, context)
464 _check_occurrence_time_validity(start_datetime, end_datetime, context)
466 if request.parent_community_id:
467 parent_node = session.execute(
468 select(Node).where(Node.id == request.parent_community_id)
469 ).scalar_one_or_none()
471 if not parent_node or not parent_node.official_cluster.small_community_features_enabled: 471 ↛ 472line 471 didn't jump to line 472 because the condition on line 471 was never true
472 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "events_not_enabled")
473 else:
474 # parent community computed from geom
475 parent_node = get_parent_node_at_location(session, not_none(geom))
477 if not parent_node: 477 ↛ 478line 477 didn't jump to line 478 because the condition on line 477 was never true
478 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "community_not_found")
480 if (
481 request.photo_key
482 and not session.execute(select(Upload).where(Upload.key == request.photo_key)).scalar_one_or_none()
483 ):
484 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "photo_not_found")
486 thread = Thread()
487 session.add(thread)
488 session.flush()
490 event = Event(
491 title=request.title,
492 parent_node_id=parent_node.id,
493 owner_user_id=context.user_id,
494 thread_id=thread.id,
495 creator_user_id=context.user_id,
496 )
497 session.add(event)
498 session.flush()
500 occurrence: EventOccurrence | None = None
502 def create_occurrence(moderation_state_id: int) -> int:
503 nonlocal occurrence
504 occurrence = EventOccurrence(
505 event_id=event.id,
506 content=request.content,
507 geom=geom,
508 address=address,
509 timezone=timezone.key,
510 photo_key=request.photo_key if request.photo_key != "" else None,
511 during=TimestamptzRange(start_datetime, end_datetime),
512 creator_user_id=context.user_id,
513 moderation_state_id=moderation_state_id,
514 )
515 session.add(occurrence)
516 session.flush()
517 return occurrence.id
519 create_moderation(
520 session=session,
521 object_type=ModerationObjectType.event_occurrence,
522 object_id=create_occurrence,
523 creator_user_id=context.user_id,
524 )
526 assert occurrence is not None
528 session.add(
529 EventOrganizer(
530 user_id=context.user_id,
531 event_id=event.id,
532 )
533 )
535 session.add(
536 EventSubscription(
537 user_id=context.user_id,
538 event_id=event.id,
539 )
540 )
542 session.add(
543 EventOccurrenceAttendee(
544 user_id=context.user_id,
545 occurrence_id=occurrence.id,
546 attendee_status=AttendeeStatus.going,
547 )
548 )
550 session.commit()
552 log_event(
553 context,
554 session,
555 "event.created",
556 {
557 "event_id": event.id,
558 "occurrence_id": occurrence.id,
559 "parent_community_id": parent_node.id,
560 "parent_community_name": parent_node.official_cluster.name,
561 },
562 )
564 if has_completed_profile(session, user): 564 ↛ 575line 564 didn't jump to line 575 because the condition on line 564 was always true
565 queue_job(
566 session,
567 job=generate_event_create_notifications,
568 payload=jobs_pb2.GenerateEventCreateNotificationsPayload(
569 inviting_user_id=user.id,
570 occurrence_id=occurrence.id,
571 approved=False,
572 ),
573 )
575 return event_to_pb(session, occurrence, context)
577 def ScheduleEvent(
578 self, request: events_pb2.ScheduleEventReq, context: CouchersContext, session: Session
579 ) -> events_pb2.Event:
580 if not request.content: 580 ↛ 581line 580 didn't jump to line 581 because the condition on line 580 was never true
581 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "missing_event_content")
583 geom, address = _check_location(request.location if request.HasField("location") else None, context)
584 timezone = _check_timezone_at(geom, context, session)
585 start_datetime = _check_iso8601_local_datetime(request.start_datetime_iso8601_local, timezone, context)
586 end_datetime = _check_iso8601_local_datetime(request.end_datetime_iso8601_local, timezone, context)
587 _check_occurrence_time_validity(start_datetime, end_datetime, context)
589 res = _get_event_and_occurrence_one_or_none(session, occurrence_id=request.event_id, context=context)
590 if not res: 590 ↛ 591line 590 didn't jump to line 591 because the condition on line 590 was never true
591 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "event_not_found")
593 event, occurrence = res
595 if not _can_edit_event(session, event, context.user_id): 595 ↛ 596line 595 didn't jump to line 596 because the condition on line 595 was never true
596 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "event_edit_permission_denied")
598 if occurrence.is_cancelled: 598 ↛ 599line 598 didn't jump to line 599 because the condition on line 598 was never true
599 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "event_cant_update_cancelled_event")
601 if ( 601 ↛ 605line 601 didn't jump to line 605 because the condition on line 601 was never true
602 request.photo_key
603 and not session.execute(select(Upload).where(Upload.key == request.photo_key)).scalar_one_or_none()
604 ):
605 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "photo_not_found")
607 during = TimestamptzRange(start_datetime, end_datetime)
609 # && is the overlap operator for ranges
610 if (
611 session.execute(
612 select(EventOccurrence.id)
613 .where(EventOccurrence.event_id == event.id)
614 .where(EventOccurrence.during.op("&&")(during))
615 .limit(1)
616 )
617 .scalars()
618 .one_or_none()
619 is not None
620 ):
621 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "event_cant_overlap")
623 new_occurrence: EventOccurrence | None = None
625 def create_occurrence(moderation_state_id: int) -> int:
626 nonlocal new_occurrence
627 new_occurrence = EventOccurrence(
628 event_id=event.id,
629 content=request.content,
630 geom=geom,
631 address=address,
632 timezone=timezone.key,
633 photo_key=request.photo_key if request.photo_key != "" else None,
634 during=during,
635 creator_user_id=context.user_id,
636 moderation_state_id=moderation_state_id,
637 )
638 session.add(new_occurrence)
639 session.flush()
640 return new_occurrence.id
642 create_moderation(
643 session=session,
644 object_type=ModerationObjectType.event_occurrence,
645 object_id=create_occurrence,
646 creator_user_id=context.user_id,
647 )
649 assert new_occurrence is not None
651 session.add(
652 EventOccurrenceAttendee(
653 user_id=context.user_id,
654 occurrence_id=new_occurrence.id,
655 attendee_status=AttendeeStatus.going,
656 )
657 )
659 session.flush()
661 # TODO: notify
663 return event_to_pb(session, new_occurrence, context)
665 def UpdateEvent(
666 self, request: events_pb2.UpdateEventReq, context: CouchersContext, session: Session
667 ) -> events_pb2.Event:
668 user = session.execute(select(User).where(User.id == context.user_id)).scalar_one()
669 res = _get_event_and_occurrence_one_or_none(session, occurrence_id=request.event_id, context=context)
670 if not res: 670 ↛ 671line 670 didn't jump to line 671 because the condition on line 670 was never true
671 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "event_not_found")
673 event, occurrence = res
675 if not _can_edit_event(session, event, context.user_id): 675 ↛ 676line 675 didn't jump to line 676 because the condition on line 675 was never true
676 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "event_edit_permission_denied")
678 # the things that were updated and need to be notified about
679 notify_updated: list[notification_data_pb2.EventUpdateItem.ValueType] = []
681 if occurrence.is_cancelled:
682 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "event_cant_update_cancelled_event")
684 occurrence_update: dict[str, Any] = {"last_edited": now()}
686 if request.HasField("title"):
687 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_TITLE)
688 event.title = request.title.value
690 if request.HasField("content"):
691 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_CONTENT)
692 occurrence_update["content"] = request.content.value
694 if request.HasField("photo_key"): 694 ↛ 695line 694 didn't jump to line 695 because the condition on line 694 was never true
695 occurrence_update["photo_key"] = request.photo_key.value
697 old_timezone = ZoneInfo(occurrence.timezone)
698 timezone: ZoneInfo = old_timezone
699 if request.HasField("location"):
700 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_LOCATION)
701 geom, address = _check_location(request.location, context)
702 timezone = _check_timezone_at(geom, context, session)
703 occurrence_update["geom"] = geom
704 occurrence_update["address"] = address
705 occurrence_update["timezone"] = timezone.key
707 if timezone != old_timezone and request.update_all_future: 707 ↛ 709line 707 didn't jump to line 709 because the condition on line 707 was never true
708 # Not implemented: We'd need to change and recheck the datetimes on all existing occurrences
709 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "event_cant_update_all_times")
711 # Determine the new start/end datetimes, which may have changed explicitly or because of a timezone change
712 start_datetime = _update_datetime(
713 request.start_datetime_iso8601_local.value if request.HasField("start_datetime_iso8601_local") else None,
714 timezone,
715 old_datetime=occurrence.start_time,
716 old_timezone=old_timezone,
717 context=context,
718 )
719 if start_datetime != occurrence.start_time:
720 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_START_TIME)
722 end_datetime = _update_datetime(
723 request.end_datetime_iso8601_local.value if request.HasField("end_datetime_iso8601_local") else None,
724 timezone,
725 old_datetime=occurrence.end_time,
726 old_timezone=old_timezone,
727 context=context,
728 )
729 if end_datetime != occurrence.end_time:
730 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_END_TIME)
732 if start_datetime != occurrence.start_time or end_datetime != occurrence.end_time:
733 _check_occurrence_time_validity(start_datetime, end_datetime, context)
735 during = TimestamptzRange(start_datetime, end_datetime)
737 # && is the overlap operator for ranges
738 if (
739 session.execute(
740 select(EventOccurrence.id)
741 .where(EventOccurrence.event_id == event.id)
742 .where(EventOccurrence.id != occurrence.id)
743 .where(EventOccurrence.during.op("&&")(during))
744 .limit(1)
745 )
746 .scalars()
747 .one_or_none()
748 is not None
749 ):
750 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "event_cant_overlap")
752 occurrence_update["during"] = during
754 # allow editing any event which hasn't ended more than 24 hours before now
755 # when editing all future events, we edit all which have not yet ended
757 cutoff_time = now() - timedelta(hours=24)
758 if request.update_all_future:
759 session.execute(
760 update(EventOccurrence)
761 .where(EventOccurrence.end_time >= cutoff_time)
762 .where(EventOccurrence.start_time >= occurrence.start_time)
763 .values(occurrence_update)
764 .execution_options(synchronize_session=False)
765 )
766 else:
767 if occurrence.end_time < cutoff_time: 767 ↛ 768line 767 didn't jump to line 768 because the condition on line 767 was never true
768 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "event_cant_update_old_event")
769 session.execute(
770 update(EventOccurrence)
771 .where(EventOccurrence.end_time >= cutoff_time)
772 .where(EventOccurrence.id == occurrence.id)
773 .values(occurrence_update)
774 .execution_options(synchronize_session=False)
775 )
777 session.flush()
779 if notify_updated:
780 items_str = ",".join(notification_data_pb2.EventUpdateItem.Name(item) for item in notify_updated)
781 if request.should_notify:
782 logger.info(f"Items {items_str} updated in event {event.id=}, notifying")
784 queue_job(
785 session,
786 job=generate_event_update_notifications,
787 payload=jobs_pb2.GenerateEventUpdateNotificationsPayload(
788 updating_user_id=user.id,
789 occurrence_id=occurrence.id,
790 updated_enum_items=notify_updated,
791 ),
792 )
793 else:
794 logger.info(f"Items {items_str} updated in event {event.id=}, but skipping notifications")
796 # since we have synchronize_session=False, we have to refresh the object
797 session.refresh(occurrence)
799 return event_to_pb(session, occurrence, context)
801 def GetEvent(self, request: events_pb2.GetEventReq, context: CouchersContext, session: Session) -> events_pb2.Event:
802 query = select(EventOccurrence).where(EventOccurrence.id == request.event_id).where(~EventOccurrence.is_deleted)
803 query = where_moderated_content_visible(query, context, EventOccurrence, is_list_operation=False)
804 occurrence = session.execute(query).scalar_one_or_none()
806 if not occurrence:
807 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "event_not_found")
809 return event_to_pb(session, occurrence, context)
811 def CancelEvent(
812 self, request: events_pb2.CancelEventReq, context: CouchersContext, session: Session
813 ) -> empty_pb2.Empty:
814 res = _get_event_and_occurrence_one_or_none(session, occurrence_id=request.event_id, context=context)
815 if not res: 815 ↛ 816line 815 didn't jump to line 816 because the condition on line 815 was never true
816 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "event_not_found")
818 event, occurrence = res
820 if not _can_edit_event(session, event, context.user_id): 820 ↛ 821line 820 didn't jump to line 821 because the condition on line 820 was never true
821 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "event_edit_permission_denied")
823 if occurrence.end_time < now() - timedelta(hours=24): 823 ↛ 824line 823 didn't jump to line 824 because the condition on line 823 was never true
824 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "event_cant_cancel_old_event")
826 occurrence.is_cancelled = True
828 log_event(context, session, "event.cancelled", {"event_id": event.id, "occurrence_id": occurrence.id})
830 queue_job(
831 session,
832 job=generate_event_cancel_notifications,
833 payload=jobs_pb2.GenerateEventCancelNotificationsPayload(
834 cancelling_user_id=context.user_id,
835 occurrence_id=occurrence.id,
836 ),
837 )
839 return empty_pb2.Empty()
841 def RequestCommunityInvite(
842 self, request: events_pb2.RequestCommunityInviteReq, context: CouchersContext, session: Session
843 ) -> empty_pb2.Empty:
844 user = session.execute(select(User).where(User.id == context.user_id)).scalar_one()
845 res = _get_event_and_occurrence_one_or_none(session, occurrence_id=request.event_id, context=context)
846 if not res: 846 ↛ 847line 846 didn't jump to line 847 because the condition on line 846 was never true
847 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "event_not_found")
849 event, occurrence = res
851 if not _can_edit_event(session, event, context.user_id):
852 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "event_edit_permission_denied")
854 if occurrence.is_cancelled: 854 ↛ 855line 854 didn't jump to line 855 because the condition on line 854 was never true
855 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "event_cant_update_cancelled_event")
857 if occurrence.end_time < now() - timedelta(hours=24): 857 ↛ 858line 857 didn't jump to line 858 because the condition on line 857 was never true
858 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "event_cant_update_old_event")
860 this_user_reqs = [req for req in occurrence.community_invite_requests if req.user_id == context.user_id]
862 if len(this_user_reqs) > 0:
863 context.abort_with_error_code(
864 grpc.StatusCode.FAILED_PRECONDITION, "event_community_invite_already_requested"
865 )
867 approved_reqs = [req for req in occurrence.community_invite_requests if req.approved]
869 if len(approved_reqs) > 0:
870 context.abort_with_error_code(
871 grpc.StatusCode.FAILED_PRECONDITION, "event_community_invite_already_approved"
872 )
874 req = EventCommunityInviteRequest(
875 occurrence_id=request.event_id,
876 user_id=context.user_id,
877 )
878 session.add(req)
879 session.flush()
881 send_event_community_invite_request_email(session, req)
883 return empty_pb2.Empty()
885 def ListEventOccurrences(
886 self, request: events_pb2.ListEventOccurrencesReq, context: CouchersContext, session: Session
887 ) -> events_pb2.ListEventOccurrencesRes:
888 page_size = min(MAX_PAGINATION_LENGTH, request.page_size or MAX_PAGINATION_LENGTH)
889 # the page token is a unix timestamp of where we left off
890 page_token = dt_from_millis(int(request.page_token)) if request.page_token else now()
891 initial_query = (
892 select(EventOccurrence).where(EventOccurrence.id == request.event_id).where(~EventOccurrence.is_deleted)
893 )
894 initial_query = where_moderated_content_visible(
895 initial_query, context, EventOccurrence, is_list_operation=False
896 )
897 occurrence = session.execute(initial_query).scalar_one_or_none()
898 if not occurrence: 898 ↛ 899line 898 didn't jump to line 899 because the condition on line 898 was never true
899 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "event_not_found")
901 query = (
902 select(EventOccurrence)
903 .where(EventOccurrence.event_id == occurrence.event_id)
904 .where(~EventOccurrence.is_deleted)
905 )
906 query = where_moderated_content_visible(query, context, EventOccurrence, is_list_operation=True)
908 if not request.include_cancelled:
909 query = query.where(~EventOccurrence.is_cancelled)
911 if not request.past: 911 ↛ 915line 911 didn't jump to line 915 because the condition on line 911 was always true
912 cutoff = page_token - timedelta(seconds=1)
913 query = query.where(EventOccurrence.end_time > cutoff).order_by(EventOccurrence.start_time.asc())
914 else:
915 cutoff = page_token + timedelta(seconds=1)
916 query = query.where(EventOccurrence.end_time < cutoff).order_by(EventOccurrence.start_time.desc())
918 query = query.limit(page_size + 1)
919 occurrences = session.execute(query).scalars().all()
921 return events_pb2.ListEventOccurrencesRes(
922 events=[event_to_pb(session, occurrence, context) for occurrence in occurrences[:page_size]],
923 next_page_token=str(millis_from_dt(occurrences[-1].end_time)) if len(occurrences) > page_size else None,
924 )
926 def ListEventAttendees(
927 self, request: events_pb2.ListEventAttendeesReq, context: CouchersContext, session: Session
928 ) -> events_pb2.ListEventAttendeesRes:
929 page_size = min(MAX_PAGINATION_LENGTH, request.page_size or MAX_PAGINATION_LENGTH)
930 next_user_id = int(request.page_token) if request.page_token else 0
931 occurrence = session.execute(
932 where_moderated_content_visible(
933 select(EventOccurrence)
934 .where(EventOccurrence.id == request.event_id)
935 .where(~EventOccurrence.is_deleted),
936 context,
937 EventOccurrence,
938 is_list_operation=False,
939 )
940 ).scalar_one_or_none()
941 if not occurrence: 941 ↛ 942line 941 didn't jump to line 942 because the condition on line 941 was never true
942 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "event_not_found")
943 attendees = (
944 session.execute(
945 where_users_column_visible(
946 select(EventOccurrenceAttendee)
947 .where(EventOccurrenceAttendee.occurrence_id == occurrence.id)
948 .where(EventOccurrenceAttendee.user_id >= next_user_id)
949 .order_by(EventOccurrenceAttendee.user_id)
950 .limit(page_size + 1),
951 context,
952 EventOccurrenceAttendee.user_id,
953 )
954 )
955 .scalars()
956 .all()
957 )
958 return events_pb2.ListEventAttendeesRes(
959 attendee_user_ids=[attendee.user_id for attendee in attendees[:page_size]],
960 next_page_token=str(attendees[-1].user_id) if len(attendees) > page_size else None,
961 )
963 def ListEventSubscribers(
964 self, request: events_pb2.ListEventSubscribersReq, context: CouchersContext, session: Session
965 ) -> events_pb2.ListEventSubscribersRes:
966 page_size = min(MAX_PAGINATION_LENGTH, request.page_size or MAX_PAGINATION_LENGTH)
967 next_user_id = int(request.page_token) if request.page_token else 0
968 res = _get_event_and_occurrence_one_or_none(session, occurrence_id=request.event_id, context=context)
969 if not res: 969 ↛ 970line 969 didn't jump to line 970 because the condition on line 969 was never true
970 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "event_not_found")
971 event, occurrence = res
972 subscribers = (
973 session.execute(
974 where_users_column_visible(
975 select(EventSubscription)
976 .where(EventSubscription.event_id == event.id)
977 .where(EventSubscription.user_id >= next_user_id)
978 .order_by(EventSubscription.user_id)
979 .limit(page_size + 1),
980 context,
981 EventSubscription.user_id,
982 )
983 )
984 .scalars()
985 .all()
986 )
987 return events_pb2.ListEventSubscribersRes(
988 subscriber_user_ids=[subscriber.user_id for subscriber in subscribers[:page_size]],
989 next_page_token=str(subscribers[-1].user_id) if len(subscribers) > page_size else None,
990 )
992 def ListEventOrganizers(
993 self, request: events_pb2.ListEventOrganizersReq, context: CouchersContext, session: Session
994 ) -> events_pb2.ListEventOrganizersRes:
995 page_size = min(MAX_PAGINATION_LENGTH, request.page_size or MAX_PAGINATION_LENGTH)
996 next_user_id = int(request.page_token) if request.page_token else 0
997 res = _get_event_and_occurrence_one_or_none(session, occurrence_id=request.event_id, context=context)
998 if not res: 998 ↛ 999line 998 didn't jump to line 999 because the condition on line 998 was never true
999 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "event_not_found")
1000 event, occurrence = res
1001 organizers = (
1002 session.execute(
1003 where_users_column_visible(
1004 select(EventOrganizer)
1005 .where(EventOrganizer.event_id == event.id)
1006 .where(EventOrganizer.user_id >= next_user_id)
1007 .order_by(EventOrganizer.user_id)
1008 .limit(page_size + 1),
1009 context,
1010 EventOrganizer.user_id,
1011 )
1012 )
1013 .scalars()
1014 .all()
1015 )
1016 return events_pb2.ListEventOrganizersRes(
1017 organizer_user_ids=[organizer.user_id for organizer in organizers[:page_size]],
1018 next_page_token=str(organizers[-1].user_id) if len(organizers) > page_size else None,
1019 )
1021 def TransferEvent(
1022 self, request: events_pb2.TransferEventReq, context: CouchersContext, session: Session
1023 ) -> events_pb2.Event:
1024 res = _get_event_and_occurrence_one_or_none(session, occurrence_id=request.event_id, context=context)
1025 if not res: 1025 ↛ 1026line 1025 didn't jump to line 1026 because the condition on line 1025 was never true
1026 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "event_not_found")
1028 event, occurrence = res
1030 if not _can_edit_event(session, event, context.user_id):
1031 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "event_transfer_permission_denied")
1033 if occurrence.is_cancelled:
1034 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "event_cant_update_cancelled_event")
1036 if occurrence.end_time < now() - timedelta(hours=24): 1036 ↛ 1037line 1036 didn't jump to line 1037 because the condition on line 1036 was never true
1037 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "event_cant_update_old_event")
1039 if request.WhichOneof("new_owner") == "new_owner_group_id":
1040 cluster = session.execute(
1041 select(Cluster).where(~Cluster.is_official_cluster).where(Cluster.id == request.new_owner_group_id)
1042 ).scalar_one_or_none()
1043 elif request.WhichOneof("new_owner") == "new_owner_community_id": 1043 ↛ 1050line 1043 didn't jump to line 1050 because the condition on line 1043 was always true
1044 cluster = session.execute(
1045 select(Cluster)
1046 .where(Cluster.parent_node_id == request.new_owner_community_id)
1047 .where(Cluster.is_official_cluster)
1048 ).scalar_one_or_none()
1050 if not cluster: 1050 ↛ 1051line 1050 didn't jump to line 1051 because the condition on line 1050 was never true
1051 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "group_or_community_not_found")
1053 event.owner_user = None
1054 event.owner_cluster = cluster
1056 session.commit()
1057 return event_to_pb(session, occurrence, context)
1059 def SetEventSubscription(
1060 self, request: events_pb2.SetEventSubscriptionReq, context: CouchersContext, session: Session
1061 ) -> events_pb2.Event:
1062 res = _get_event_and_occurrence_one_or_none(session, occurrence_id=request.event_id, context=context)
1063 if not res: 1063 ↛ 1064line 1063 didn't jump to line 1064 because the condition on line 1063 was never true
1064 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "event_not_found")
1066 event, occurrence = res
1068 if occurrence.is_cancelled:
1069 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "event_cant_update_cancelled_event")
1071 if occurrence.end_time < now() - timedelta(hours=24): 1071 ↛ 1072line 1071 didn't jump to line 1072 because the condition on line 1071 was never true
1072 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "event_cant_update_old_event")
1074 current_subscription = session.execute(
1075 select(EventSubscription)
1076 .where(EventSubscription.user_id == context.user_id)
1077 .where(EventSubscription.event_id == event.id)
1078 ).scalar_one_or_none()
1080 # if not subscribed, subscribe
1081 if request.subscribe and not current_subscription:
1082 session.add(EventSubscription(user_id=context.user_id, event_id=event.id))
1084 # if subscribed but unsubbing, remove subscription
1085 if not request.subscribe and current_subscription:
1086 session.delete(current_subscription)
1088 session.flush()
1090 log_event(
1091 context,
1092 session,
1093 "event.subscription_set",
1094 {"event_id": event.id, "occurrence_id": occurrence.id, "subscribed": request.subscribe},
1095 )
1097 return event_to_pb(session, occurrence, context)
1099 def SetEventAttendance(
1100 self, request: events_pb2.SetEventAttendanceReq, context: CouchersContext, session: Session
1101 ) -> events_pb2.Event:
1102 occurrence = session.execute(
1103 where_moderated_content_visible(
1104 select(EventOccurrence)
1105 .where(EventOccurrence.id == request.event_id)
1106 .where(~EventOccurrence.is_deleted),
1107 context,
1108 EventOccurrence,
1109 is_list_operation=False,
1110 )
1111 ).scalar_one_or_none()
1113 if not occurrence: 1113 ↛ 1114line 1113 didn't jump to line 1114 because the condition on line 1113 was never true
1114 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "event_not_found")
1116 if occurrence.is_cancelled:
1117 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "event_cant_update_cancelled_event")
1119 if occurrence.end_time < now() - timedelta(hours=24): 1119 ↛ 1120line 1119 didn't jump to line 1120 because the condition on line 1119 was never true
1120 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "event_cant_update_old_event")
1122 current_attendance = session.execute(
1123 select(EventOccurrenceAttendee)
1124 .where(EventOccurrenceAttendee.user_id == context.user_id)
1125 .where(EventOccurrenceAttendee.occurrence_id == occurrence.id)
1126 ).scalar_one_or_none()
1128 if request.attendance_state == events_pb2.ATTENDANCE_STATE_NOT_GOING:
1129 if current_attendance: 1129 ↛ 1144line 1129 didn't jump to line 1144 because the condition on line 1129 was always true
1130 session.delete(current_attendance)
1131 # if unset/not going, nothing to do!
1132 else:
1133 if current_attendance: 1133 ↛ 1134line 1133 didn't jump to line 1134 because the condition on line 1133 was never true
1134 current_attendance.attendee_status = attendancestate2sql[request.attendance_state] # type: ignore[assignment]
1135 else:
1136 # create new
1137 attendance = EventOccurrenceAttendee(
1138 user_id=context.user_id,
1139 occurrence_id=occurrence.id,
1140 attendee_status=not_none(attendancestate2sql[request.attendance_state]),
1141 )
1142 session.add(attendance)
1144 session.flush()
1146 log_event(
1147 context,
1148 session,
1149 "event.attendance_set",
1150 {"occurrence_id": occurrence.id, "attendance_state": request.attendance_state},
1151 )
1153 return event_to_pb(session, occurrence, context)
1155 def ListMyEvents(
1156 self, request: events_pb2.ListMyEventsReq, context: CouchersContext, session: Session
1157 ) -> events_pb2.ListMyEventsRes:
1158 page_size = min(MAX_PAGINATION_LENGTH, request.page_size or MAX_PAGINATION_LENGTH)
1159 # the page token is a unix timestamp of where we left off
1160 page_token = (
1161 dt_from_millis(int(request.page_token)) if request.page_token and not request.page_number else now()
1162 )
1163 # the page number is the page number we are on
1164 page_number = request.page_number or 1
1165 # Calculate the offset for pagination
1166 offset = (page_number - 1) * page_size
1167 query = (
1168 select(EventOccurrence).join(Event, Event.id == EventOccurrence.event_id).where(~EventOccurrence.is_deleted)
1169 )
1170 query = where_moderated_content_visible(query, context, EventOccurrence, is_list_operation=True)
1172 include_all = not (request.subscribed or request.attending or request.organizing or request.my_communities)
1173 include_subscribed = request.subscribed or include_all
1174 include_organizing = request.organizing or include_all
1175 include_attending = request.attending or include_all
1176 include_my_communities = request.my_communities or include_all
1178 if include_attending and request.exclude_attending:
1179 context.abort_with_error_code(
1180 grpc.StatusCode.INVALID_ARGUMENT, "cannot_combine_attending_and_exclude_attending"
1181 )
1183 where_ = []
1185 if include_subscribed:
1186 query = query.outerjoin(
1187 EventSubscription,
1188 and_(EventSubscription.event_id == Event.id, EventSubscription.user_id == context.user_id),
1189 )
1190 where_.append(EventSubscription.user_id != None)
1191 if include_organizing:
1192 query = query.outerjoin(
1193 EventOrganizer, and_(EventOrganizer.event_id == Event.id, EventOrganizer.user_id == context.user_id)
1194 )
1195 where_.append(EventOrganizer.user_id != None)
1196 if include_attending or request.exclude_attending:
1197 query = query.outerjoin(
1198 EventOccurrenceAttendee,
1199 and_(
1200 EventOccurrenceAttendee.occurrence_id == EventOccurrence.id,
1201 EventOccurrenceAttendee.user_id == context.user_id,
1202 ),
1203 )
1204 if include_attending:
1205 where_.append(EventOccurrenceAttendee.user_id != None)
1206 elif request.exclude_attending: 1206 ↛ 1213line 1206 didn't jump to line 1213 because the condition on line 1206 was always true
1207 if not include_organizing: 1207 ↛ 1212line 1207 didn't jump to line 1212 because the condition on line 1207 was always true
1208 query = query.outerjoin(
1209 EventOrganizer,
1210 and_(EventOrganizer.event_id == Event.id, EventOrganizer.user_id == context.user_id),
1211 )
1212 query = query.where(EventOccurrenceAttendee.user_id == None, EventOrganizer.user_id == None)
1213 if include_my_communities:
1214 my_communities = (
1215 session.execute(
1216 select(Node.id)
1217 .join(Cluster, Cluster.parent_node_id == Node.id)
1218 .join(ClusterSubscription, ClusterSubscription.cluster_id == Cluster.id)
1219 .where(ClusterSubscription.user_id == context.user_id)
1220 .where(Cluster.is_official_cluster)
1221 .order_by(Node.id)
1222 .limit(100000)
1223 )
1224 .scalars()
1225 .all()
1226 )
1227 where_.append(Event.parent_node_id.in_(my_communities))
1229 query = query.where(or_(*where_))
1231 if request.my_communities_exclude_global:
1232 query = query.join(Node, Node.id == Event.parent_node_id).where(Node.node_type > NodeType.region)
1234 if not request.include_cancelled:
1235 query = query.where(~EventOccurrence.is_cancelled)
1237 if not request.past: 1237 ↛ 1241line 1237 didn't jump to line 1241 because the condition on line 1237 was always true
1238 cutoff = page_token - timedelta(seconds=1)
1239 query = query.where(EventOccurrence.end_time > cutoff).order_by(EventOccurrence.start_time.asc())
1240 else:
1241 cutoff = page_token + timedelta(seconds=1)
1242 query = query.where(EventOccurrence.end_time < cutoff).order_by(EventOccurrence.start_time.desc())
1243 # Count the total number of items for pagination
1244 total_items = session.execute(select(func.count()).select_from(query.subquery())).scalar()
1245 # Apply pagination by page number
1246 query = query.offset(offset).limit(page_size) if request.page_number else query.limit(page_size + 1)
1247 occurrences = session.execute(query).scalars().all()
1249 return events_pb2.ListMyEventsRes(
1250 events=[event_to_pb(session, occurrence, context) for occurrence in occurrences[:page_size]],
1251 next_page_token=str(millis_from_dt(occurrences[-1].end_time)) if len(occurrences) > page_size else None,
1252 total_items=total_items,
1253 )
1255 def ListAllEvents(
1256 self, request: events_pb2.ListAllEventsReq, context: CouchersContext, session: Session
1257 ) -> events_pb2.ListAllEventsRes:
1258 page_size = min(MAX_PAGINATION_LENGTH, request.page_size or MAX_PAGINATION_LENGTH)
1259 # the page token is a unix timestamp of where we left off
1260 page_token = dt_from_millis(int(request.page_token)) if request.page_token else now()
1262 query = select(EventOccurrence).where(~EventOccurrence.is_deleted)
1263 query = where_moderated_content_visible(query, context, EventOccurrence, is_list_operation=True)
1265 if not request.include_cancelled: 1265 ↛ 1268line 1265 didn't jump to line 1268 because the condition on line 1265 was always true
1266 query = query.where(~EventOccurrence.is_cancelled)
1268 if not request.past:
1269 cutoff = page_token - timedelta(seconds=1)
1270 query = query.where(EventOccurrence.end_time > cutoff).order_by(EventOccurrence.start_time.asc())
1271 else:
1272 cutoff = page_token + timedelta(seconds=1)
1273 query = query.where(EventOccurrence.end_time < cutoff).order_by(EventOccurrence.start_time.desc())
1275 query = query.limit(page_size + 1)
1276 occurrences = session.execute(query).scalars().all()
1278 return events_pb2.ListAllEventsRes(
1279 events=[event_to_pb(session, occurrence, context) for occurrence in occurrences[:page_size]],
1280 next_page_token=str(millis_from_dt(occurrences[-1].end_time)) if len(occurrences) > page_size else None,
1281 )
1283 def InviteEventOrganizer(
1284 self, request: events_pb2.InviteEventOrganizerReq, context: CouchersContext, session: Session
1285 ) -> empty_pb2.Empty:
1286 user = session.execute(select(User).where(User.id == context.user_id)).scalar_one()
1287 res = _get_event_and_occurrence_one_or_none(session, occurrence_id=request.event_id, context=context)
1288 if not res: 1288 ↛ 1289line 1288 didn't jump to line 1289 because the condition on line 1288 was never true
1289 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "event_not_found")
1291 event, occurrence = res
1293 if not _can_edit_event(session, event, context.user_id):
1294 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "event_edit_permission_denied")
1296 if occurrence.is_cancelled:
1297 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "event_cant_update_cancelled_event")
1299 if occurrence.end_time < now() - timedelta(hours=24): 1299 ↛ 1300line 1299 didn't jump to line 1300 because the condition on line 1299 was never true
1300 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "event_cant_update_old_event")
1302 if not session.execute( 1302 ↛ 1305line 1302 didn't jump to line 1305 because the condition on line 1302 was never true
1303 select(User).where(users_visible(context)).where(User.id == request.user_id)
1304 ).scalar_one_or_none():
1305 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
1307 session.add(
1308 EventOrganizer(
1309 user_id=request.user_id,
1310 event_id=event.id,
1311 )
1312 )
1313 session.flush()
1315 other_user_context = make_notification_user_context(user_id=request.user_id)
1317 notify(
1318 session,
1319 user_id=request.user_id,
1320 topic_action=NotificationTopicAction.event__invite_organizer,
1321 key=str(event.id),
1322 data=notification_data_pb2.EventInviteOrganizer(
1323 event=event_to_pb(session, occurrence, other_user_context),
1324 inviting_user=user_model_to_pb(user, session, other_user_context),
1325 ),
1326 )
1328 return empty_pb2.Empty()
1330 def RemoveEventOrganizer(
1331 self, request: events_pb2.RemoveEventOrganizerReq, context: CouchersContext, session: Session
1332 ) -> empty_pb2.Empty:
1333 res = _get_event_and_occurrence_one_or_none(session, occurrence_id=request.event_id, context=context)
1334 if not res: 1334 ↛ 1335line 1334 didn't jump to line 1335 because the condition on line 1334 was never true
1335 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "event_not_found")
1337 event, occurrence = res
1339 if occurrence.is_cancelled: 1339 ↛ 1340line 1339 didn't jump to line 1340 because the condition on line 1339 was never true
1340 context.abort_with_error_code(grpc.StatusCode.PERMISSION_DENIED, "event_cant_update_cancelled_event")
1342 if occurrence.end_time < now() - timedelta(hours=24): 1342 ↛ 1343line 1342 didn't jump to line 1343 because the condition on line 1342 was never true
1343 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "event_cant_update_old_event")
1345 # Determine which user to remove
1346 user_id_to_remove = request.user_id.value if request.HasField("user_id") else context.user_id
1348 # Check if the target user is the event owner (only after permission check)
1349 if event.owner_user_id == user_id_to_remove:
1350 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "event_cant_remove_owner_as_organizer")
1352 # Check permissions: either an organizer removing an organizer OR you're the event owner
1353 if not _can_edit_event(session, event, context.user_id):
1354 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "event_edit_permission_denied")
1356 # Find the organizer to remove
1357 organizer_to_remove = session.execute(
1358 select(EventOrganizer)
1359 .where(EventOrganizer.user_id == user_id_to_remove)
1360 .where(EventOrganizer.event_id == event.id)
1361 ).scalar_one_or_none()
1363 if not organizer_to_remove: 1363 ↛ 1364line 1363 didn't jump to line 1364 because the condition on line 1363 was never true
1364 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "event_not_an_organizer")
1366 session.delete(organizer_to_remove)
1368 return empty_pb2.Empty()