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 notify

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 _can_moderate_event(session: Session, event: Event, user_id: int) -> bool:

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

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

101 return True

102

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

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

105

106

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

108 return (

109 _is_event_owner(event, user_id)

110 or _is_event_organizer(event, user_id)

111 or _can_moderate_event(session, event, user_id)

112 )

113

114

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

116 event = occurrence.event

117

118 next_occurrence = (

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

120 .order_by(EventOccurrence.end_time.asc())

121 .limit(1)

122 .one_or_none()

123 )

124

125 owner_community_id = None

126 owner_group_id = None

127 if event.owner_cluster:

128 if event.owner_cluster.is_official_cluster:

129 owner_community_id = event.owner_cluster.parent_node_id

130 else:

131 owner_group_id = event.owner_cluster.id

132

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

134 attendance_state = attendance.attendee_status if attendance else None

135

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

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

138

139 going_count = session.execute(

140 where_users_column_visible(

141 select(func.count())

142 .select_from(EventOccurrenceAttendee)

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

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

145 context,

146 EventOccurrenceAttendee.user_id,

147 )

148 ).scalar_one()

149 organizer_count = session.execute(

150 where_users_column_visible(

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

152 context,

153 EventOrganizer.user_id,

154 )

155 ).scalar_one()

156 subscriber_count = session.execute(

157 where_users_column_visible(

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

159 context,

160 EventSubscription.user_id,

161 )

162 ).scalar_one()

163

164 return events_pb2.Event(

165 event_id=occurrence.id,

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

167 is_cancelled=occurrence.is_cancelled,

168 is_deleted=occurrence.is_deleted,

169 title=event.title,

170 slug=event.slug,

171 content=occurrence.content,

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

173 photo_key=occurrence.photo_key or "",

174 location=events_pb2.EventLocation(

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

176 ),

177 created=Timestamp_from_datetime(occurrence.created),

178 last_edited=Timestamp_from_datetime(occurrence.last_edited),

179 creator_user_id=occurrence.creator_user_id,

180 start_time=Timestamp_from_datetime(occurrence.start_time),

181 end_time=Timestamp_from_datetime(occurrence.end_time),

182 timezone=occurrence.timezone,

183 attendance_state=attendancestate2api[attendance_state],

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

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

186 going_count=going_count,

187 organizer_count=organizer_count,

188 subscriber_count=subscriber_count,

189 owner_user_id=event.owner_user_id,

190 owner_community_id=owner_community_id,

191 owner_group_id=owner_group_id,

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

193 can_edit=can_edit,

194 can_moderate=can_moderate,

195 )

196

197

198def _get_event_and_occurrence_query(

199 occurrence_id: int,

200 include_deleted: bool,

201 context: CouchersContext | None = None,

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

203 query = (

204 select(Event, EventOccurrence)

205 .where(EventOccurrence.id == occurrence_id)

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

207 )

208

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

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

211

212 if context is not None:

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

214

215 return query

216

217

218def _get_event_and_occurrence_one(

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

220) -> tuple[Event, EventOccurrence]:

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

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

223 return result._tuple()

224

225

226def _get_event_and_occurrence_one_or_none(

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

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

229 result = session.execute(

230 _get_event_and_occurrence_query(occurrence_id, include_deleted, context=context)

231 ).one_or_none()

232 return result._tuple() if result else None

233

234

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

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

237 if not location or not location.address:

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

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

240 # No events allowed on Null Island

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

242

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

244 return (geom, location.address)

245

246

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

248 timezone_id = session.execute(

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

250 ).scalar_one_or_none()

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

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

253

254 return ZoneInfo(timezone_id)

255

256

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

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

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

260

261 try:

262 naive_datetime = datetime.fromisoformat(value)

263 except ValueError:

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

265

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

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

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

269

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

271

272

273def _update_datetime(

274 new_iso8601_local: str | None,

275 new_timezone: ZoneInfo,

276 old_datetime: datetime,

277 old_timezone: ZoneInfo,

278 context: CouchersContext,

279) -> datetime:

280 if new_iso8601_local is None and new_timezone != old_timezone:

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

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

283 if new_iso8601_local is None:

284 return old_datetime # No change

285 # New effective datetime/timestamp

286 return _check_iso8601_local_datetime(new_iso8601_local, new_timezone, context)

287

288

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

290 if start_time < now():

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

292 if end_time < start_time:

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

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

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

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

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

298

299

300def apply_occurrence_pagination(

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

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

303 """

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

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

306 """

307 if not past:

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

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

310 )

311 if page_token:

312 start_time, occurrence_id = dt_id_from_page_token(page_token)

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

314 else:

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

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

317 )

318 if page_token:

319 start_time, occurrence_id = dt_id_from_page_token(page_token)

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

321 return query

322

323

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

325 if len(occurrences) <= page_size:

326 return None

327 next_occurrence = occurrences[page_size]

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

329

330

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

332 """

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

334 """

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

336 not_already_involved = User.id.not_in(

337 select(EventOccurrenceAttendee.user_id)

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

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

340 )

341

342 cluster = occurrence.event.parent_node.official_cluster

343 creator = aliased(User)

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

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

346 return [], occurrence.event.parent_node_id

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

348 members = (

349 session.execute(

350 select(User)

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

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

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

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

355 .where(not_already_involved)

356 )

357 .scalars()

358 .all()

359 )

360 return list(members), occurrence.event.parent_node_id

361 else:

362 max_radius = 20000 # m

363 users = (

364 session.execute(

365 select(User)

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

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

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

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

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

371 .where(not_already_involved)

372 )

373 .scalars()

374 .all()

375 )

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

377

378

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

380 """

381 Background job to generated/fan out event notifications

382 """

383 # Import here to avoid circular dependency

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

385

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

387

388 with session_scope() as session:

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

390

391 users, node_id = get_users_to_notify_for_new_event(session, occurrence)

392

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

394

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

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

397 return

398

399 for user in users:

400 context = make_notification_user_context(user_id=user.id)

401 topic_action = (

402 NotificationTopicAction.event__create_approved

403 if payload.approved

404 else NotificationTopicAction.event__create_any

405 )

406 notify(

407 session,

408 user_id=user.id,

409 topic_action=topic_action,

410 key=str(payload.occurrence_id),

411 data=notification_data_pb2.EventCreate(

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

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

414 nearby=True if node_id is None else None,

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

416 ),

417 moderation_state_id=occurrence.moderation_state_id,

418 )

419

420

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

422 with session_scope() as session:

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

424

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

426

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

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

429

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

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

432 continue

433 context = make_notification_user_context(user_id=user_id)

434 notify(

435 session,

436 user_id=user_id,

437 topic_action=NotificationTopicAction.event__update,

438 key=str(payload.occurrence_id),

439 data=notification_data_pb2.EventUpdate(

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

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

442 updated_enum_items=(

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

444 ),

445 ),

446 moderation_state_id=occurrence.moderation_state_id,

447 )

448

449

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

451 with session_scope() as session:

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

453

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

455

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

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

458

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

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

461 continue

462 context = make_notification_user_context(user_id=user_id)

463 notify(

464 session,

465 user_id=user_id,

466 topic_action=NotificationTopicAction.event__cancel,

467 key=str(payload.occurrence_id),

468 data=notification_data_pb2.EventCancel(

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

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

471 ),

472 moderation_state_id=occurrence.moderation_state_id,

473 )

474

475

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

477 with session_scope() as session:

478 event, occurrence = _get_event_and_occurrence_one(

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

480 )

481

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

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

484

485 for user_id in set(subscribed_user_ids + attending_user_ids):

486 context = make_notification_user_context(user_id=user_id)

487 notify(

488 session,

489 user_id=user_id,

490 topic_action=NotificationTopicAction.event__delete,

491 key=str(payload.occurrence_id),

492 data=notification_data_pb2.EventDelete(

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

494 ),

495 moderation_state_id=occurrence.moderation_state_id,

496 )

497

498

499class Events(events_pb2_grpc.EventsServicer):

500 def CreateEvent(

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

502 ) -> events_pb2.Event:

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

504 if not has_completed_profile(session, user):

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

506 if not request.title:

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

508 if not request.content:

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

510

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

512 timezone = _check_timezone_at(geom, context, session)

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

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

515 _check_occurrence_time_validity(start_datetime, end_datetime, context)

516

517 if request.parent_community_id:

518 parent_node = session.execute(

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

520 ).scalar_one_or_none()

521

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

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

524 else:

525 # parent community computed from geom

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

527

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

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

530

531 if (

532 request.photo_key

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

534 ):

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

536

537 event = Event(

538 title=request.title,

539 parent_node_id=parent_node.id,

540 owner_user_id=context.user_id,

541 creator_user_id=context.user_id,

542 )

543 session.add(event)

544 session.flush()

545

546 thread = Thread()

547 session.add(thread)

548 session.flush()

549

550 occurrence: EventOccurrence | None = None

551

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

553 nonlocal occurrence

554 occurrence = EventOccurrence(

555 event_id=event.id,

556 content=request.content,

557 geom=geom,

558 address=address,

559 timezone=timezone.key,

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

561 during=TimestamptzRange(start_datetime, end_datetime),

562 creator_user_id=context.user_id,

563 moderation_state_id=moderation_state_id,

564 thread_id=thread.id,

565 )

566 session.add(occurrence)

567 session.flush()

568 return occurrence.id

569

570 create_moderation(

571 session=session,

572 object_type=ModerationObjectType.event_occurrence,

573 object_id=create_occurrence,

574 creator_user_id=context.user_id,

575 )

576

577 assert occurrence is not None

578

579 session.add(

580 EventOrganizer(

581 user_id=context.user_id,

582 event_id=event.id,

583 )

584 )

585

586 session.add(

587 EventSubscription(

588 user_id=context.user_id,

589 event_id=event.id,

590 )

591 )

592

593 session.add(

594 EventOccurrenceAttendee(

595 user_id=context.user_id,

596 occurrence_id=occurrence.id,

597 attendee_status=AttendeeStatus.going,

598 )

599 )

600

601 session.commit()

602

603 log_event(

604 context,

605 session,

606 "event.created",

607 {

608 "event_id": event.id,

609 "occurrence_id": occurrence.id,

610 "parent_community_id": parent_node.id,

611 "parent_community_name": parent_node.official_cluster.name,

612 },

613 )

614

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

616 queue_job(

617 session,

618 job=generate_event_create_notifications,

619 payload=jobs_pb2.GenerateEventCreateNotificationsPayload(

620 inviting_user_id=user.id,

621 occurrence_id=occurrence.id,

622 approved=False,

623 ),

624 )

625

626 return event_to_pb(session, occurrence, context)

627

628 def ScheduleEvent(

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

630 ) -> events_pb2.Event:

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

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

633

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

635 timezone = _check_timezone_at(geom, context, session)

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

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

638 _check_occurrence_time_validity(start_datetime, end_datetime, context)

639

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

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

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

643

644 event, occurrence = res

645

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

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

648

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

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

651

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

653 request.photo_key

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

655 ):

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

657

658 during = TimestamptzRange(start_datetime, end_datetime)

659

660 # && is the overlap operator for ranges

661 if (

662 session.execute(

663 select(EventOccurrence.id)

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

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

666 .limit(1)

667 )

668 .scalars()

669 .one_or_none()

670 is not None

671 ):

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

673

674 thread = Thread()

675 session.add(thread)

676 session.flush()

677

678 new_occurrence: EventOccurrence | None = None

679

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

681 nonlocal new_occurrence

682 new_occurrence = EventOccurrence(

683 event_id=event.id,

684 content=request.content,

685 geom=geom,

686 address=address,

687 timezone=timezone.key,

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

689 during=during,

690 creator_user_id=context.user_id,

691 moderation_state_id=moderation_state_id,

692 thread_id=thread.id,

693 )

694 session.add(new_occurrence)

695 session.flush()

696 return new_occurrence.id

697

698 create_moderation(

699 session=session,

700 object_type=ModerationObjectType.event_occurrence,

701 object_id=create_occurrence,

702 creator_user_id=context.user_id,

703 )

704

705 assert new_occurrence is not None

706

707 session.add(

708 EventOccurrenceAttendee(

709 user_id=context.user_id,

710 occurrence_id=new_occurrence.id,

711 attendee_status=AttendeeStatus.going,

712 )

713 )

714

715 session.flush()

716

717 # TODO: notify

718

719 return event_to_pb(session, new_occurrence, context)

720

721 def UpdateEvent(

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

723 ) -> events_pb2.Event:

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

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

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

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

728

729 event, occurrence = res

730

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

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

733

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

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

736

737 if occurrence.is_cancelled:

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

739

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

741

742 if request.HasField("title"):

743 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_TITLE)

744 event.title = request.title.value

745

746 if request.HasField("content"):

747 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_CONTENT)

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

749

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

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

752

753 old_timezone = ZoneInfo(occurrence.timezone)

754 timezone: ZoneInfo = old_timezone

755 if request.HasField("location"):

756 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_LOCATION)

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

758 timezone = _check_timezone_at(geom, context, session)

759 occurrence_update["geom"] = geom

760 occurrence_update["address"] = address

761 occurrence_update["timezone"] = timezone.key

762

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

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

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

766

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

768 start_datetime = _update_datetime(

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

770 timezone,

771 old_datetime=occurrence.start_time,

772 old_timezone=old_timezone,

773 context=context,

774 )

775 if start_datetime != occurrence.start_time:

776 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_START_TIME)

777

778 end_datetime = _update_datetime(

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

780 timezone,

781 old_datetime=occurrence.end_time,

782 old_timezone=old_timezone,

783 context=context,

784 )

785 if end_datetime != occurrence.end_time:

786 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_END_TIME)

787

788 if request.update_all_future and (

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

790 ):

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

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

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

794

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

796 _check_occurrence_time_validity(start_datetime, end_datetime, context)

797

798 during = TimestamptzRange(start_datetime, end_datetime)

799

800 # && is the overlap operator for ranges

801 if (

802 session.execute(

803 select(EventOccurrence.id)

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

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

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

807 .limit(1)

808 )

809 .scalars()

810 .one_or_none()

811 is not None

812 ):

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

814

815 occurrence_update["during"] = during

816

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

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

819

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

821 if request.update_all_future:

822 session.execute(

823 update(EventOccurrence)

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

825 .where(~EventOccurrence.is_deleted)

826 .where(EventOccurrence.end_time >= cutoff_time)

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

828 .values(occurrence_update)

829 .execution_options(synchronize_session=False)

830 )

831 else:

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

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

834 session.execute(

835 update(EventOccurrence)

836 .where(EventOccurrence.end_time >= cutoff_time)

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

838 .values(occurrence_update)

839 .execution_options(synchronize_session=False)

840 )

841

842 session.flush()

843

844 if notify_updated:

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

846 if request.should_notify:

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

848

849 queue_job(

850 session,

851 job=generate_event_update_notifications,

852 payload=jobs_pb2.GenerateEventUpdateNotificationsPayload(

853 updating_user_id=user.id,

854 occurrence_id=occurrence.id,

855 updated_enum_items=notify_updated,

856 ),

857 )

858 else:

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

860

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

862 session.refresh(occurrence)

863

864 return event_to_pb(session, occurrence, context)

865

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

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

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

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

870

871 if not occurrence:

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

873

874 return event_to_pb(session, occurrence, context)

875

876 def CancelEvent(

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

878 ) -> empty_pb2.Empty:

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

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

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

882

883 event, occurrence = res

884

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

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

887

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

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

890

891 occurrence.is_cancelled = True

892

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

894

895 queue_job(

896 session,

897 job=generate_event_cancel_notifications,

898 payload=jobs_pb2.GenerateEventCancelNotificationsPayload(

899 cancelling_user_id=context.user_id,

900 occurrence_id=occurrence.id,

901 ),

902 )

903

904 return empty_pb2.Empty()

905

906 def RequestCommunityInvite(

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

908 ) -> empty_pb2.Empty:

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

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

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

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

913

914 event, occurrence = res

915

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

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

918

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

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

921

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

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

924

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

926

927 if len(this_user_reqs) > 0:

928 context.abort_with_error_code(

929 grpc.StatusCode.FAILED_PRECONDITION, "event_community_invite_already_requested"

930 )

931

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

933

934 if len(approved_reqs) > 0:

935 context.abort_with_error_code(

936 grpc.StatusCode.FAILED_PRECONDITION, "event_community_invite_already_approved"

937 )

938

939 req = EventCommunityInviteRequest(

940 occurrence_id=request.event_id,

941 user_id=context.user_id,

942 )

943 session.add(req)

944 session.flush()

945

946 send_event_community_invite_request_email(session, req)

947

948 return empty_pb2.Empty()

949

950 def ListEventOccurrences(

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

952 ) -> events_pb2.ListEventOccurrencesRes:

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

954 initial_query = (

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

956 )

957 initial_query = where_moderated_content_visible(

958 initial_query, context, EventOccurrence, is_list_operation=False

959 )

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

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

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

963

964 query = (

965 select(EventOccurrence)

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

967 .where(~EventOccurrence.is_deleted)

968 )

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

970

971 if not request.include_cancelled:

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

973

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

975

976 query = query.limit(page_size + 1)

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

978

979 return events_pb2.ListEventOccurrencesRes(

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

981 next_page_token=occurrences_next_page_token(occurrences, page_size),

982 )

983

984 def ListEventAttendees(

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

986 ) -> events_pb2.ListEventAttendeesRes:

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

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

989 occurrence = session.execute(

990 where_moderated_content_visible(

991 select(EventOccurrence)

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

993 .where(~EventOccurrence.is_deleted),

994 context,

995 EventOccurrence,

996 is_list_operation=False,

997 )

998 ).scalar_one_or_none()

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

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

1001 attendees = (

1002 session.execute(

1003 where_users_column_visible(

1004 select(EventOccurrenceAttendee)

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

1006 .where(EventOccurrenceAttendee.user_id >= next_user_id)

1007 .order_by(EventOccurrenceAttendee.user_id)

1008 .limit(page_size + 1),

1009 context,

1010 EventOccurrenceAttendee.user_id,

1011 )

1012 )

1013 .scalars()

1014 .all()

1015 )

1016 return events_pb2.ListEventAttendeesRes(

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

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

1019 )

1020

1021 def ListEventSubscribers(

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

1023 ) -> events_pb2.ListEventSubscribersRes:

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

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

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

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

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

1029 event, occurrence = res

1030 subscribers = (

1031 session.execute(

1032 where_users_column_visible(

1033 select(EventSubscription)

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

1035 .where(EventSubscription.user_id >= next_user_id)

1036 .order_by(EventSubscription.user_id)

1037 .limit(page_size + 1),

1038 context,

1039 EventSubscription.user_id,

1040 )

1041 )

1042 .scalars()

1043 .all()

1044 )

1045 return events_pb2.ListEventSubscribersRes(

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

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

1048 )

1049

1050 def ListEventOrganizers(

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

1052 ) -> events_pb2.ListEventOrganizersRes:

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

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

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

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

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

1058 event, occurrence = res

1059 organizers = (

1060 session.execute(

1061 where_users_column_visible(

1062 select(EventOrganizer)

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

1064 .where(EventOrganizer.user_id >= next_user_id)

1065 .order_by(EventOrganizer.user_id)

1066 .limit(page_size + 1),

1067 context,

1068 EventOrganizer.user_id,

1069 )

1070 )

1071 .scalars()

1072 .all()

1073 )

1074 return events_pb2.ListEventOrganizersRes(

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

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

1077 )

1078

1079 def TransferEvent(

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

1081 ) -> events_pb2.Event:

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

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

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

1085

1086 event, occurrence = res

1087

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

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

1090

1091 if occurrence.is_cancelled:

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

1093

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

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

1096

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

1098 cluster = session.execute(

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

1100 ).scalar_one_or_none()

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

1102 cluster = session.execute(

1103 select(Cluster)

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

1105 .where(Cluster.is_official_cluster)

1106 ).scalar_one_or_none()

1107

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

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

1110

1111 event.owner_user = None

1112 event.owner_cluster = cluster

1113

1114 session.commit()

1115 return event_to_pb(session, occurrence, context)

1116

1117 def SetEventSubscription(

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

1119 ) -> events_pb2.Event:

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

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

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

1123

1124 event, occurrence = res

1125

1126 if occurrence.is_cancelled:

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

1128

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

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

1131

1132 current_subscription = session.execute(

1133 select(EventSubscription)

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

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

1136 ).scalar_one_or_none()

1137

1138 # if not subscribed, subscribe

1139 if request.subscribe and not current_subscription:

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

1141

1142 # if subscribed but unsubbing, remove subscription

1143 if not request.subscribe and current_subscription:

1144 session.delete(current_subscription)

1145

1146 session.flush()

1147

1148 log_event(

1149 context,

1150 session,

1151 "event.subscription_set",

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

1153 )

1154

1155 return event_to_pb(session, occurrence, context)

1156

1157 def SetEventAttendance(

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

1159 ) -> events_pb2.Event:

1160 occurrence = session.execute(

1161 where_moderated_content_visible(

1162 select(EventOccurrence)

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

1164 .where(~EventOccurrence.is_deleted),

1165 context,

1166 EventOccurrence,

1167 is_list_operation=False,

1168 )

1169 ).scalar_one_or_none()

1170

1171 if not occurrence: 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 if occurrence.is_cancelled:

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

1176

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

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

1179

1180 current_attendance = session.execute(

1181 select(EventOccurrenceAttendee)

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

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

1184 ).scalar_one_or_none()

1185

1186 if request.attendance_state == events_pb2.ATTENDANCE_STATE_NOT_GOING:

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

1188 session.delete(current_attendance)

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

1190 else:

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

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

1193 else:

1194 # create new

1195 attendance = EventOccurrenceAttendee(

1196 user_id=context.user_id,

1197 occurrence_id=occurrence.id,

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

1199 )

1200 session.add(attendance)

1201

1202 session.flush()

1203

1204 log_event(

1205 context,

1206 session,

1207 "event.attendance_set",

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

1209 )

1210

1211 return event_to_pb(session, occurrence, context)

1212

1213 def ListMyEvents(

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

1215 ) -> events_pb2.ListMyEventsRes:

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

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

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

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

1220 page_number = request.page_number or 1

1221 # Calculate the offset for pagination

1222 offset = (page_number - 1) * page_size

1223 query = (

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

1225 )

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

1227

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

1229 include_subscribed = request.subscribed or include_all

1230 include_organizing = request.organizing or include_all

1231 include_attending = request.attending or include_all

1232 include_my_communities = request.my_communities or include_all

1233

1234 if include_attending and request.exclude_attending:

1235 context.abort_with_error_code(

1236 grpc.StatusCode.INVALID_ARGUMENT, "cannot_combine_attending_and_exclude_attending"

1237 )

1238

1239 where_ = []

1240

1241 if include_subscribed:

1242 query = query.outerjoin(

1243 EventSubscription,

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

1245 )

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

1247 if include_organizing:

1248 query = query.outerjoin(

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

1250 )

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

1252 if include_attending or request.exclude_attending:

1253 query = query.outerjoin(

1254 EventOccurrenceAttendee,

1255 and_(

1256 EventOccurrenceAttendee.occurrence_id == EventOccurrence.id,

1257 EventOccurrenceAttendee.user_id == context.user_id,

1258 ),

1259 )

1260 if include_attending:

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

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

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

1264 query = query.outerjoin(

1265 EventOrganizer,

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

1267 )

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

1269 if include_my_communities:

1270 my_communities = (

1271 session.execute(

1272 select(Node.id)

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

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

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

1276 .where(Cluster.is_official_cluster)

1277 .order_by(Node.id)

1278 .limit(100000)

1279 )

1280 .scalars()

1281 .all()

1282 )

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

1284

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

1286

1287 if request.my_communities_exclude_global:

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

1289

1290 if not request.include_cancelled:

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

1292

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

1294 # Count the total number of items for pagination

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

1296 # Apply pagination by page number

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

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

1299

1300 return events_pb2.ListMyEventsRes(

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

1302 next_page_token=occurrences_next_page_token(occurrences, page_size),

1303 total_items=total_items,

1304 )

1305

1306 def ListAllEvents(

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

1308 ) -> events_pb2.ListAllEventsRes:

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

1310

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

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

1313

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

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

1316

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

1318

1319 query = query.limit(page_size + 1)

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

1321

1322 return events_pb2.ListAllEventsRes(

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

1324 next_page_token=occurrences_next_page_token(occurrences, page_size),

1325 )

1326

1327 def InviteEventOrganizer(

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

1329 ) -> empty_pb2.Empty:

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

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

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

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

1334

1335 event, occurrence = res

1336

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

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

1339

1340 if occurrence.is_cancelled:

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

1342

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

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

1345

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

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

1348 ).scalar_one_or_none():

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

1350

1351 session.add(

1352 EventOrganizer(

1353 user_id=request.user_id,

1354 event_id=event.id,

1355 )

1356 )

1357 session.flush()

1358

1359 other_user_context = make_notification_user_context(user_id=request.user_id)

1360

1361 notify(

1362 session,

1363 user_id=request.user_id,

1364 topic_action=NotificationTopicAction.event__invite_organizer,

1365 key=str(event.id),

1366 data=notification_data_pb2.EventInviteOrganizer(

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

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

1369 ),

1370 )

1371

1372 return empty_pb2.Empty()

1373

1374 def RemoveEventOrganizer(

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

1376 ) -> empty_pb2.Empty:

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

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

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

1380

1381 event, occurrence = res

1382

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

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

1385

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

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

1388

1389 # Determine which user to remove

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

1391

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

1393 if event.owner_user_id == user_id_to_remove:

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

1395

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

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

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

1399

1400 # Find the organizer to remove

1401 organizer_to_remove = session.execute(

1402 select(EventOrganizer)

1403 .where(EventOrganizer.user_id == user_id_to_remove)

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

1405 ).scalar_one_or_none()

1406

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

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

1409

1410 session.delete(organizer_to_remove)

1411

1412 return empty_pb2.Empty()

1413

1414 def GetEventCalendarFile(

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

1416 ) -> httpbody_pb2.HttpBody:

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

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

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

1420

1421 _, occurrence_db = res

1422

1423 event_pb = event_to_pb(session, occurrence_db, context)

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

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