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

1import logging

2from collections.abc import Sequence

3from datetime import datetime, timedelta

4from typing import Any, cast

5from zoneinfo import ZoneInfo

7import grpc

8from geoalchemy2 import WKBElement

9from google.protobuf import empty_pb2

10from psycopg.types.range import TimestamptzRange

11from sqlalchemy import Select, func, select

12from sqlalchemy.orm import Session, aliased

13from sqlalchemy.sql import and_, func, or_, tuple_, update

15from couchers.context import CouchersContext, make_notification_user_context

16from couchers.db import can_moderate_node, get_parent_node_at_location, session_scope

17from couchers.email.calendar_events import create_event_ics_calendar

18from couchers.event_log import log_event

19from couchers.helpers.completed_profile import has_completed_profile

20from couchers.jobs.enqueue import queue_job

21from couchers.models import (

22 AttendeeStatus,

23 Cluster,

24 ClusterSubscription,

25 Event,

26 EventCommunityInviteRequest,

27 EventOccurrence,

28 EventOccurrenceAttendee,

29 EventOrganizer,

30 EventSubscription,

31 ModerationObjectType,

32 Node,

33 NodeType,

34 Thread,

35 Upload,

36 User,

37)

38from couchers.models.notifications import NotificationTopicAction

39from couchers.models.static import TimezoneArea

40from couchers.moderation.utils import create_moderation

41from couchers.notifications.notify import PendingNotification, notify, notify_many

42from couchers.proto import events_pb2, events_pb2_grpc, notification_data_pb2

43from couchers.proto.google.api import httpbody_pb2

44from couchers.proto.internal import jobs_pb2

45from couchers.sentry import report_message

46from couchers.servicers.api import user_model_to_pb

47from couchers.servicers.blocking import is_not_visible

48from couchers.servicers.threads import thread_to_pb

49from couchers.sql import (

50 users_visible,

51 users_visible_to_each_other,

52 where_moderated_content_visible,

53 where_users_column_visible,

54)

55from couchers.tasks import send_event_community_invite_request_email

56from couchers.utils import (

57 Timestamp_from_datetime,

58 create_coordinate,

59 datetime_to_iso8601_local,

60 dt_id_from_page_token,

61 dt_id_to_page_token,

62 not_none,

63 now,

64)

66logger = logging.getLogger(__name__)

68attendancestate2sql = {

69 events_pb2.AttendanceState.ATTENDANCE_STATE_NOT_GOING: None,

70 events_pb2.AttendanceState.ATTENDANCE_STATE_GOING: AttendeeStatus.going,

71}

73attendancestate2api = {

74 None: events_pb2.AttendanceState.ATTENDANCE_STATE_NOT_GOING,

75 AttendeeStatus.going: events_pb2.AttendanceState.ATTENDANCE_STATE_GOING,

76}

78MAX_PAGINATION_LENGTH = 25

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

82 """

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

84 """

85 if event.owner_user:

86 return event.owner_user_id == user_id

87 # otherwise owned by a cluster

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

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

92 """

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

94 """

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

98def _community_invite_requested(session: Session, occurrence: EventOccurrence, user_id: int) -> bool:

100 # Returns True if the given user has already requested a community invite

101 # for this event occurrence, or if one has already been approved.

102 return (

103 session.execute(

104 select(EventCommunityInviteRequest.id)

105 .where(EventCommunityInviteRequest.occurrence_id == occurrence.id)

106 .where(EventCommunityInviteRequest.user_id == user_id)

107 .limit(1)

108 ).scalar_one_or_none()

109 is not None

110 )

111

112

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

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

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

116 return True

117

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

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

120

121

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

123 return (

124 _is_event_owner(event, user_id)

125 or _is_event_organizer(event, user_id)

126 or _can_moderate_event(session, event, user_id)

127 )

128

129

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

131 event = occurrence.event

132

133 next_occurrence = (

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

135 .order_by(EventOccurrence.end_time.asc())

136 .limit(1)

137 .one_or_none()

138 )

139

140 owner_community_id = None

141 owner_group_id = None

142 if event.owner_cluster:

143 if event.owner_cluster.is_official_cluster:

144 owner_community_id = event.owner_cluster.parent_node_id

145 else:

146 owner_group_id = event.owner_cluster.id

147

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

149 attendance_state = attendance.attendee_status if attendance else None

150

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

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

153 community_invite_requested = (

154 _community_invite_requested(session, occurrence, context.user_id) if can_edit else False

155 )

156

157 going_count = session.execute(

158 where_users_column_visible(

159 select(func.count())

160 .select_from(EventOccurrenceAttendee)

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

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

163 context,

164 EventOccurrenceAttendee.user_id,

165 )

166 ).scalar_one()

167 organizer_count = session.execute(

168 where_users_column_visible(

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

170 context,

171 EventOrganizer.user_id,

172 )

173 ).scalar_one()

174 subscriber_count = session.execute(

175 where_users_column_visible(

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

177 context,

178 EventSubscription.user_id,

179 )

180 ).scalar_one()

181

182 return events_pb2.Event(

183 event_id=occurrence.id,

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

185 is_cancelled=occurrence.is_cancelled,

186 is_deleted=occurrence.is_deleted,

187 title=event.title,

188 slug=event.slug,

189 content=occurrence.content,

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

191 photo_key=occurrence.photo_key or "",

192 location=events_pb2.EventLocation(

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

194 ),

195 created=Timestamp_from_datetime(occurrence.created),

196 last_edited=Timestamp_from_datetime(occurrence.last_edited),

197 creator_user_id=occurrence.creator_user_id,

198 start_time=Timestamp_from_datetime(occurrence.start_time),

199 end_time=Timestamp_from_datetime(occurrence.end_time),

200 timezone=occurrence.timezone,

201 attendance_state=attendancestate2api[attendance_state],

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

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

204 going_count=going_count,

205 organizer_count=organizer_count,

206 subscriber_count=subscriber_count,

207 owner_user_id=event.owner_user_id,

208 owner_community_id=owner_community_id,

209 owner_group_id=owner_group_id,

210 thread=thread_to_pb(session, context, occurrence.thread_id),

211 can_edit=can_edit,

212 can_moderate=can_moderate,

213 community_invite_requested=community_invite_requested,

214 )

215

216

217def _get_event_and_occurrence_query(

218 occurrence_id: int,

219 include_deleted: bool,

220 context: CouchersContext | None = None,

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

222 query = (

223 select(Event, EventOccurrence)

224 .where(EventOccurrence.id == occurrence_id)

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

226 )

227

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

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

230

231 if context is not None:

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

233

234 return query

235

236

237def _get_event_and_occurrence_one(

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

239) -> tuple[Event, EventOccurrence]:

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

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

242 return result._tuple()

243

244

245def _get_event_and_occurrence_one_or_none(

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

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

248 result = session.execute(

249 _get_event_and_occurrence_query(occurrence_id, include_deleted, context=context)

250 ).one_or_none()

251 return result._tuple() if result else None

252

253

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

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

256 if not location or not location.address:

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

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

259 # No events allowed on Null Island

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

261

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

263 return (geom, location.address)

264

265

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

267 timezone_id = session.execute(

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

269 ).scalar_one_or_none()

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

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

272

273 return ZoneInfo(timezone_id)

274

275

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

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

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

279

280 try:

281 naive_datetime = datetime.fromisoformat(value)

282 except ValueError:

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

284

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

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

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

288

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

290

291

292def _update_datetime(

293 new_iso8601_local: str | None,

294 new_timezone: ZoneInfo,

295 old_datetime: datetime,

296 old_timezone: ZoneInfo,

297 context: CouchersContext,

298) -> datetime:

299 if new_iso8601_local is None and new_timezone != old_timezone:

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

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

302 if new_iso8601_local is None:

303 return old_datetime # No change

304 # New effective datetime/timestamp

305 return _check_iso8601_local_datetime(new_iso8601_local, new_timezone, context)

306

307

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

309 if start_time < now():

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

311 if end_time < start_time:

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

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

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

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

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

317

318

319def apply_occurrence_pagination(

320 query: Select[tuple[EventOccurrence]], page_token: str, past: bool

321) -> Select[tuple[EventOccurrence]]:

322 """

323 Restricts to upcoming (not yet ended) or past occurrences, orders by start time (with id as

324 tiebreaker), and seeks to the (start_time, id) cursor from the page token, if given.

325 """

326 if not past:

327 query = query.where(EventOccurrence.end_time > now()).order_by(

328 EventOccurrence.start_time.asc(), EventOccurrence.id.asc()

329 )

330 if page_token:

331 start_time, occurrence_id = dt_id_from_page_token(page_token)

332 query = query.where(tuple_(EventOccurrence.start_time, EventOccurrence.id) >= (start_time, occurrence_id))

333 else:

334 query = query.where(EventOccurrence.end_time < now()).order_by(

335 EventOccurrence.start_time.desc(), EventOccurrence.id.desc()

336 )

337 if page_token:

338 start_time, occurrence_id = dt_id_from_page_token(page_token)

339 query = query.where(tuple_(EventOccurrence.start_time, EventOccurrence.id) <= (start_time, occurrence_id))

340 return query

341

342

343def occurrences_next_page_token(occurrences: Sequence[EventOccurrence], page_size: int) -> str | None:

344 if len(occurrences) <= page_size:

345 return None

346 next_occurrence = occurrences[page_size]

347 return dt_id_to_page_token(next_occurrence.start_time, next_occurrence.id)

348

349

350def is_community_too_large_for_event_notifications(node: Node) -> bool:

351 return node.node_type.value <= NodeType.region.value

352

353

354def get_users_to_notify_for_new_event(

355 session: Session, occurrence: EventOccurrence, node: Node | None = None

356) -> tuple[list[User], int | None]:

357 """

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

359

360 Defaults to the event's current parent community.

361 """

362 node = node or occurrence.event.parent_node

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

364 not_already_involved = User.id.not_in(

365 select(EventOccurrenceAttendee.user_id)

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

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

368 )

369

370 cluster = node.official_cluster

371 creator = aliased(User)

372 if is_community_too_large_for_event_notifications(node):

373 logger.info(f"Community {node.id} is a {node.node_type} and too big for email notifications.")

374 return [], node.id

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

376 members = (

377 session.execute(

378 select(User)

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

380 .join_from(User, creator, creator.id == occurrence.creator_user_id)

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

382 .where(users_visible_to_each_other(self_user=User, other_user=creator))

383 .where(not_already_involved)

384 )

385 .scalars()

386 .all()

387 )

388 return list(members), node.id

389 else:

390 max_radius = 20000 # m

391 users = (

392 session.execute(

393 select(User)

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

395 .join_from(User, creator, creator.id == occurrence.creator_user_id)

396 .where(users_visible_to_each_other(self_user=User, other_user=creator))

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

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

399 .where(not_already_involved)

400 )

401 .scalars()

402 .all()

403 )

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

405

406

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

408 """

409 Background job to generated/fan out event notifications

410 """

411 # Import here to avoid circular dependency

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

413

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

415

416 notifications = []

417 with session_scope() as session:

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

419

420 users, node_id = get_users_to_notify_for_new_event(session, occurrence)

421

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

423

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

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

426 return

427

428 for user in users:

429 context = make_notification_user_context(user_id=user.id)

430 topic_action = (

431 NotificationTopicAction.event__create_approved

432 if payload.approved

433 else NotificationTopicAction.event__create_any

434 )

435 notifications.append(

436 PendingNotification(

437 user_id=user.id,

438 topic_action=topic_action,

439 key=str(payload.occurrence_id),

440 data=notification_data_pb2.EventCreate(

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

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

443 nearby=True if node_id is None else None,

444 in_community=community_to_pb(session, event.parent_node, context)

445 if node_id is not None

446 else None,

447 ),

448 moderation_state_id=occurrence.moderation_state_id,

449 )

450 )

451

452 with session_scope() as session:

453 notify_many(session, notifications)

454

455

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

457 notifications = []

458 with session_scope() as session:

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

460

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

462

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

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

465

466 for user_id in set(subscribed_user_ids + attending_user_ids) - {updating_user.id}:

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

468 continue

469 context = make_notification_user_context(user_id=user_id)

470 notifications.append(

471 PendingNotification(

472 user_id=user_id,

473 topic_action=NotificationTopicAction.event__update,

474 key=str(payload.occurrence_id),

475 data=notification_data_pb2.EventUpdate(

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

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

478 updated_enum_items=(

479 notification_data_pb2.EventUpdateItem.ValueType(value)

480 for value in payload.updated_enum_items

481 ),

482 ),

483 moderation_state_id=occurrence.moderation_state_id,

484 )

485 )

486

487 with session_scope() as session:

488 notify_many(session, notifications)

489

490

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

492 notifications = []

493 with session_scope() as session:

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

495

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

497

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

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

500

501 for user_id in set(subscribed_user_ids + attending_user_ids) - {cancelling_user.id}:

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

503 continue

504 context = make_notification_user_context(user_id=user_id)

505 notifications.append(

506 PendingNotification(

507 user_id=user_id,

508 topic_action=NotificationTopicAction.event__cancel,

509 key=str(payload.occurrence_id),

510 data=notification_data_pb2.EventCancel(

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

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

513 ),

514 moderation_state_id=occurrence.moderation_state_id,

515 )

516 )

517

518 with session_scope() as session:

519 notify_many(session, notifications)

520

521

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

523 notifications = []

524 with session_scope() as session:

525 event, occurrence = _get_event_and_occurrence_one(

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

527 )

528

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

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

531

532 for user_id in set(subscribed_user_ids + attending_user_ids):

533 context = make_notification_user_context(user_id=user_id)

534 notifications.append(

535 PendingNotification(

536 user_id=user_id,

537 topic_action=NotificationTopicAction.event__delete,

538 key=str(payload.occurrence_id),

539 data=notification_data_pb2.EventDelete(

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

541 ),

542 moderation_state_id=occurrence.moderation_state_id,

543 )

544 )

545

546 with session_scope() as session:

547 notify_many(session, notifications)

548

549

550class Events(events_pb2_grpc.EventsServicer):

551 def CreateEvent(

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

553 ) -> events_pb2.Event:

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

555 if not has_completed_profile(session, user):

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

557 if not request.title:

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

559 if not request.content:

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

561

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

563 timezone = _check_timezone_at(geom, context, session)

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

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

566 _check_occurrence_time_validity(start_datetime, end_datetime, context)

567

568 if request.parent_community_id:

569 parent_node = session.execute(

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

571 ).scalar_one_or_none()

572

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

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

575 else:

576 # parent community computed from geom

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

578

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

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

581

582 if (

583 request.photo_key

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

585 ):

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

587

588 event = Event(

589 title=request.title,

590 parent_node_id=parent_node.id,

591 owner_user_id=context.user_id,

592 creator_user_id=context.user_id,

593 )

594 session.add(event)

595 session.flush()

596

597 thread = Thread()

598 session.add(thread)

599 session.flush()

600

601 occurrence: EventOccurrence | None = None

602

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

604 nonlocal occurrence

605 occurrence = EventOccurrence(

606 event_id=event.id,

607 content=request.content,

608 geom=geom,

609 address=address,

610 timezone=timezone.key,

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

612 during=TimestamptzRange(start_datetime, end_datetime),

613 creator_user_id=context.user_id,

614 moderation_state_id=moderation_state_id,

615 thread_id=thread.id,

616 )

617 session.add(occurrence)

618 session.flush()

619 return occurrence.id

620

621 create_moderation(

622 session=session,

623 object_type=ModerationObjectType.event_occurrence,

624 object_id=create_occurrence,

625 creator_user_id=context.user_id,

626 )

627

628 assert occurrence is not None

629

630 session.add(

631 EventOrganizer(

632 user_id=context.user_id,

633 event_id=event.id,

634 )

635 )

636

637 session.add(

638 EventSubscription(

639 user_id=context.user_id,

640 event_id=event.id,

641 )

642 )

643

644 session.add(

645 EventOccurrenceAttendee(

646 user_id=context.user_id,

647 occurrence_id=occurrence.id,

648 attendee_status=AttendeeStatus.going,

649 )

650 )

651

652 session.commit()

653

654 log_event(

655 context,

656 session,

657 "event.created",

658 {

659 "event_id": event.id,

660 "occurrence_id": occurrence.id,

661 "parent_community_id": parent_node.id,

662 "parent_community_name": parent_node.official_cluster.name,

663 },

664 )

665

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

667 queue_job(

668 session,

669 job=generate_event_create_notifications,

670 payload=jobs_pb2.GenerateEventCreateNotificationsPayload(

671 inviting_user_id=user.id,

672 occurrence_id=occurrence.id,

673 approved=False,

674 ),

675 )

676

677 return event_to_pb(session, occurrence, context)

678

679 def ScheduleEvent(

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

681 ) -> events_pb2.Event:

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

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

684

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

686 timezone = _check_timezone_at(geom, context, session)

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

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

689 _check_occurrence_time_validity(start_datetime, end_datetime, context)

690

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

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

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

694

695 event, occurrence = res

696

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

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

699

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

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

702

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

704 request.photo_key

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

706 ):

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

708

709 during = TimestamptzRange(start_datetime, end_datetime)

710

711 # && is the overlap operator for ranges

712 if (

713 session.execute(

714 select(EventOccurrence.id)

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

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

717 .limit(1)

718 )

719 .scalars()

720 .one_or_none()

721 is not None

722 ):

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

724

725 thread = Thread()

726 session.add(thread)

727 session.flush()

728

729 new_occurrence: EventOccurrence | None = None

730

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

732 nonlocal new_occurrence

733 new_occurrence = EventOccurrence(

734 event_id=event.id,

735 content=request.content,

736 geom=geom,

737 address=address,

738 timezone=timezone.key,

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

740 during=during,

741 creator_user_id=context.user_id,

742 moderation_state_id=moderation_state_id,

743 thread_id=thread.id,

744 )

745 session.add(new_occurrence)

746 session.flush()

747 return new_occurrence.id

748

749 create_moderation(

750 session=session,

751 object_type=ModerationObjectType.event_occurrence,

752 object_id=create_occurrence,

753 creator_user_id=context.user_id,

754 )

755

756 assert new_occurrence is not None

757

758 session.add(

759 EventOccurrenceAttendee(

760 user_id=context.user_id,

761 occurrence_id=new_occurrence.id,

762 attendee_status=AttendeeStatus.going,

763 )

764 )

765

766 session.flush()

767

768 # TODO: notify

769

770 return event_to_pb(session, new_occurrence, context)

771

772 def UpdateEvent(

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

774 ) -> events_pb2.Event:

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

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

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

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

779

780 event, occurrence = res

781

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

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

784

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

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

787

788 if occurrence.is_cancelled:

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

790

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

792

793 if request.HasField("title"):

794 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_TITLE)

795 event.title = request.title.value

796

797 if request.HasField("content"):

798 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_CONTENT)

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

800

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

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

803

804 old_timezone = ZoneInfo(occurrence.timezone)

805 timezone: ZoneInfo = old_timezone

806 if request.HasField("location"):

807 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_LOCATION)

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

809 timezone = _check_timezone_at(geom, context, session)

810 occurrence_update["geom"] = geom

811 occurrence_update["address"] = address

812 occurrence_update["timezone"] = timezone.key

813

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

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

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

817

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

819 start_datetime = _update_datetime(

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

821 timezone,

822 old_datetime=occurrence.start_time,

823 old_timezone=old_timezone,

824 context=context,

825 )

826 if start_datetime != occurrence.start_time:

827 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_START_TIME)

828

829 end_datetime = _update_datetime(

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

831 timezone,

832 old_datetime=occurrence.end_time,

833 old_timezone=old_timezone,

834 context=context,

835 )

836 if end_datetime != occurrence.end_time:

837 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_END_TIME)

838

839 if request.update_all_future and (

840 start_datetime != occurrence.start_time or end_datetime != occurrence.end_time

841 ):

842 # Not implemented: every future occurrence would need its own times, rechecked against the others.

843 # Writing this one range to all of them violates the exclusion constraint on (event_id, during).

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

845

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

847 _check_occurrence_time_validity(start_datetime, end_datetime, context)

848

849 during = TimestamptzRange(start_datetime, end_datetime)

850

851 # && is the overlap operator for ranges

852 if (

853 session.execute(

854 select(EventOccurrence.id)

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

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

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

858 .limit(1)

859 )

860 .scalars()

861 .one_or_none()

862 is not None

863 ):

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

865

866 occurrence_update["during"] = during

867

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

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

870

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

872 if request.update_all_future:

873 session.execute(

874 update(EventOccurrence)

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

876 .where(~EventOccurrence.is_deleted)

877 .where(EventOccurrence.end_time >= cutoff_time)

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

879 .values(occurrence_update)

880 .execution_options(synchronize_session=False)

881 )

882 else:

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

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

885 session.execute(

886 update(EventOccurrence)

887 .where(EventOccurrence.end_time >= cutoff_time)

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

889 .values(occurrence_update)

890 .execution_options(synchronize_session=False)

891 )

892

893 session.flush()

894

895 if notify_updated:

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

897 if request.should_notify:

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

899

900 queue_job(

901 session,

902 job=generate_event_update_notifications,

903 payload=jobs_pb2.GenerateEventUpdateNotificationsPayload(

904 updating_user_id=user.id,

905 occurrence_id=occurrence.id,

906 updated_enum_items=notify_updated,

907 ),

908 )

909 else:

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

911

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

913 session.refresh(occurrence)

914

915 return event_to_pb(session, occurrence, context)

916

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

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

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

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

921

922 if not occurrence:

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

924

925 return event_to_pb(session, occurrence, context)

926

927 def CancelEvent(

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

929 ) -> empty_pb2.Empty:

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

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

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

933

934 event, occurrence = res

935

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

938

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

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

941

942 occurrence.is_cancelled = True

943

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

945

946 queue_job(

947 session,

948 job=generate_event_cancel_notifications,

949 payload=jobs_pb2.GenerateEventCancelNotificationsPayload(

950 cancelling_user_id=context.user_id,

951 occurrence_id=occurrence.id,

952 ),

953 )

954

955 return empty_pb2.Empty()

956

957 def RequestCommunityInvite(

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

959 ) -> empty_pb2.Empty:

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

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

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

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

964

965 event, occurrence = res

966

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

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

969

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

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

972

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

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

975

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

977 if len(this_user_reqs) > 0:

978 context.abort_with_error_code(

979 grpc.StatusCode.FAILED_PRECONDITION, "event_community_invite_already_requested"

980 )

981

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

983

984 if len(approved_reqs) > 0:

985 context.abort_with_error_code(

986 grpc.StatusCode.FAILED_PRECONDITION, "event_community_invite_already_approved"

987 )

988

989 req = EventCommunityInviteRequest(

990 occurrence_id=request.event_id,

991 user_id=context.user_id,

992 )

993 session.add(req)

994 session.flush()

995

996 send_event_community_invite_request_email(session, req)

997

998 return empty_pb2.Empty()

999

1000 def ListEventOccurrences(

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

1002 ) -> events_pb2.ListEventOccurrencesRes:

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

1004 initial_query = (

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

1006 )

1007 initial_query = where_moderated_content_visible(

1008 initial_query, context, EventOccurrence, is_list_operation=False

1009 )

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

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

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

1013

1014 query = (

1015 select(EventOccurrence)

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

1017 .where(~EventOccurrence.is_deleted)

1018 )

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

1020

1021 if not request.include_cancelled:

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

1023

1024 query = apply_occurrence_pagination(query, request.page_token, request.past)

1025

1026 query = query.limit(page_size + 1)

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

1028

1029 return events_pb2.ListEventOccurrencesRes(

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

1031 next_page_token=occurrences_next_page_token(occurrences, page_size),

1032 )

1033

1034 def ListEventAttendees(

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

1036 ) -> events_pb2.ListEventAttendeesRes:

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

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

1039 occurrence = session.execute(

1040 where_moderated_content_visible(

1041 select(EventOccurrence)

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

1043 .where(~EventOccurrence.is_deleted),

1044 context,

1045 EventOccurrence,

1046 is_list_operation=False,

1047 )

1048 ).scalar_one_or_none()

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

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

1051 attendees = (

1052 session.execute(

1053 where_users_column_visible(

1054 select(EventOccurrenceAttendee)

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

1056 .where(EventOccurrenceAttendee.user_id >= next_user_id)

1057 .order_by(EventOccurrenceAttendee.user_id)

1058 .limit(page_size + 1),

1059 context,

1060 EventOccurrenceAttendee.user_id,

1061 )

1062 )

1063 .scalars()

1064 .all()

1065 )

1066 return events_pb2.ListEventAttendeesRes(

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

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

1069 )

1070

1071 def ListEventSubscribers(

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

1073 ) -> events_pb2.ListEventSubscribersRes:

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

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

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

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

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

1079 event, occurrence = res

1080 subscribers = (

1081 session.execute(

1082 where_users_column_visible(

1083 select(EventSubscription)

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

1085 .where(EventSubscription.user_id >= next_user_id)

1086 .order_by(EventSubscription.user_id)

1087 .limit(page_size + 1),

1088 context,

1089 EventSubscription.user_id,

1090 )

1091 )

1092 .scalars()

1093 .all()

1094 )

1095 return events_pb2.ListEventSubscribersRes(

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

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

1098 )

1099

1100 def ListEventOrganizers(

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

1102 ) -> events_pb2.ListEventOrganizersRes:

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

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

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

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

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

1108 event, occurrence = res

1109 organizers = (

1110 session.execute(

1111 where_users_column_visible(

1112 select(EventOrganizer)

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

1114 .where(EventOrganizer.user_id >= next_user_id)

1115 .order_by(EventOrganizer.user_id)

1116 .limit(page_size + 1),

1117 context,

1118 EventOrganizer.user_id,

1119 )

1120 )

1121 .scalars()

1122 .all()

1123 )

1124 return events_pb2.ListEventOrganizersRes(

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

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

1127 )

1128

1129 def TransferEvent(

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

1131 ) -> events_pb2.Event:

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

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

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

1135

1136 event, occurrence = res

1137

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

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

1140

1141 if occurrence.is_cancelled:

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

1143

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

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

1146

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

1148 cluster = session.execute(

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

1150 ).scalar_one_or_none()

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

1152 cluster = session.execute(

1153 select(Cluster)

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

1155 .where(Cluster.is_official_cluster)

1156 ).scalar_one_or_none()

1157

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

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

1160

1161 event.owner_user = None

1162 event.owner_cluster = cluster

1163

1164 session.commit()

1165 return event_to_pb(session, occurrence, context)

1166

1167 def SetEventSubscription(

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

1169 ) -> events_pb2.Event:

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

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

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

1173

1174 event, occurrence = res

1175

1176 if occurrence.is_cancelled:

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

1178

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

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

1181

1182 current_subscription = session.execute(

1183 select(EventSubscription)

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

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

1186 ).scalar_one_or_none()

1187

1188 # if not subscribed, subscribe

1189 if request.subscribe and not current_subscription:

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

1191

1192 # if subscribed but unsubbing, remove subscription

1193 if not request.subscribe and current_subscription:

1194 session.delete(current_subscription)

1195

1196 session.flush()

1197

1198 log_event(

1199 context,

1200 session,

1201 "event.subscription_set",

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

1203 )

1204

1205 return event_to_pb(session, occurrence, context)

1206

1207 def SetEventAttendance(

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

1209 ) -> events_pb2.Event:

1210 occurrence = session.execute(

1211 where_moderated_content_visible(

1212 select(EventOccurrence)

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

1214 .where(~EventOccurrence.is_deleted),

1215 context,

1216 EventOccurrence,

1217 is_list_operation=False,

1218 )

1219 ).scalar_one_or_none()

1220

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

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

1223

1224 if occurrence.is_cancelled:

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

1226

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

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

1229

1230 current_attendance = session.execute(

1231 select(EventOccurrenceAttendee)

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

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

1234 ).scalar_one_or_none()

1235

1236 if request.attendance_state == events_pb2.ATTENDANCE_STATE_NOT_GOING:

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

1238 session.delete(current_attendance)

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

1240 else:

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

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

1243 else:

1244 # create new

1245 attendance = EventOccurrenceAttendee(

1246 user_id=context.user_id,

1247 occurrence_id=occurrence.id,

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

1249 )

1250 session.add(attendance)

1251

1252 session.flush()

1253

1254 log_event(

1255 context,

1256 session,

1257 "event.attendance_set",

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

1259 )

1260

1261 return event_to_pb(session, occurrence, context)

1262

1263 def ListMyEvents(

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

1265 ) -> events_pb2.ListMyEventsRes:

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

1267 # the page token is ignored when a page number is given

1268 page_token = request.page_token if not request.page_number else ""

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

1270 page_number = request.page_number or 1

1271 # Calculate the offset for pagination

1272 offset = (page_number - 1) * page_size

1273 query = (

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

1275 )

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

1277

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

1279 include_subscribed = request.subscribed or include_all

1280 include_organizing = request.organizing or include_all

1281 include_attending = request.attending or include_all

1282 include_my_communities = request.my_communities or include_all

1283

1284 if include_attending and request.exclude_attending:

1285 context.abort_with_error_code(

1286 grpc.StatusCode.INVALID_ARGUMENT, "cannot_combine_attending_and_exclude_attending"

1287 )

1288

1289 where_ = []

1290

1291 if include_subscribed:

1292 query = query.outerjoin(

1293 EventSubscription,

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

1295 )

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

1297 if include_organizing:

1298 query = query.outerjoin(

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

1300 )

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

1302 if include_attending or request.exclude_attending:

1303 query = query.outerjoin(

1304 EventOccurrenceAttendee,

1305 and_(

1306 EventOccurrenceAttendee.occurrence_id == EventOccurrence.id,

1307 EventOccurrenceAttendee.user_id == context.user_id,

1308 ),

1309 )

1310 if include_attending:

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

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

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

1314 query = query.outerjoin(

1315 EventOrganizer,

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

1317 )

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

1319 if include_my_communities:

1320 my_communities = (

1321 session.execute(

1322 select(Node.id)

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

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

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

1326 .where(Cluster.is_official_cluster)

1327 .order_by(Node.id)

1328 .limit(100000)

1329 )

1330 .scalars()

1331 .all()

1332 )

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

1334

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

1336

1337 if request.my_communities_exclude_global:

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

1339

1340 if not request.include_cancelled:

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

1342

1343 query = apply_occurrence_pagination(query, page_token, request.past)

1344 # Count the total number of items for pagination

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

1346 # Apply pagination by page number

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

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

1349

1350 return events_pb2.ListMyEventsRes(

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

1352 next_page_token=occurrences_next_page_token(occurrences, page_size),

1353 total_items=total_items,

1354 )

1355

1356 def ListAllEvents(

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

1358 ) -> events_pb2.ListAllEventsRes:

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

1360

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

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

1363

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

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

1366

1367 query = apply_occurrence_pagination(query, request.page_token, request.past)

1368

1369 query = query.limit(page_size + 1)

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

1371

1372 return events_pb2.ListAllEventsRes(

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

1374 next_page_token=occurrences_next_page_token(occurrences, page_size),

1375 )

1376

1377 def InviteEventOrganizer(

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

1379 ) -> empty_pb2.Empty:

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

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

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

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

1384

1385 event, occurrence = res

1386

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

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

1389

1390 if occurrence.is_cancelled:

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

1392

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

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

1395

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

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

1398 ).scalar_one_or_none():

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

1400

1401 session.add(

1402 EventOrganizer(

1403 user_id=request.user_id,

1404 event_id=event.id,

1405 )

1406 )

1407 session.flush()

1408

1409 other_user_context = make_notification_user_context(user_id=request.user_id)

1410

1411 notify(

1412 session,

1413 user_id=request.user_id,

1414 topic_action=NotificationTopicAction.event__invite_organizer,

1415 key=str(event.id),

1416 data=notification_data_pb2.EventInviteOrganizer(

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

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

1419 ),

1420 )

1421

1422 return empty_pb2.Empty()

1423

1424 def RemoveEventOrganizer(

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

1426 ) -> empty_pb2.Empty:

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

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

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

1430

1431 event, occurrence = res

1432

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

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

1435

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

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

1438

1439 # Determine which user to remove

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

1441

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

1443 if event.owner_user_id == user_id_to_remove:

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

1445

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

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

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

1449

1450 # Find the organizer to remove

1451 organizer_to_remove = session.execute(

1452 select(EventOrganizer)

1453 .where(EventOrganizer.user_id == user_id_to_remove)

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

1455 ).scalar_one_or_none()

1456

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

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

1459

1460 session.delete(organizer_to_remove)

1461

1462 return empty_pb2.Empty()

1463

1464 def GetEventCalendarFile(

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

1466 ) -> httpbody_pb2.HttpBody:

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

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

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

1470

1471 _, occurrence_db = res

1472

1473 event_pb = event_to_pb(session, occurrence_db, context)

1474 ics_data = create_event_ics_calendar(event_pb, context.localization).to_ical()

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