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.servicers.api import user_model_to_pb

46from couchers.servicers.blocking import is_not_visible

47from couchers.servicers.threads import thread_to_pb

48from couchers.sql import (

49 users_visible,

50 users_visible_to_each_other,

51 where_moderated_content_visible,

52 where_users_column_visible,

53)

54from couchers.tasks import send_event_community_invite_request_email

55from couchers.utils import (

56 Timestamp_from_datetime,

57 create_coordinate,

58 datetime_to_iso8601_local,

59 dt_id_from_page_token,

60 dt_id_to_page_token,

61 not_none,

62 now,

63)

65logger = logging.getLogger(__name__)

67attendancestate2sql = {

68 events_pb2.AttendanceState.ATTENDANCE_STATE_NOT_GOING: None,

69 events_pb2.AttendanceState.ATTENDANCE_STATE_GOING: AttendeeStatus.going,

70}

72attendancestate2api = {

73 None: events_pb2.AttendanceState.ATTENDANCE_STATE_NOT_GOING,

74 AttendeeStatus.going: events_pb2.AttendanceState.ATTENDANCE_STATE_GOING,

75}

77MAX_PAGINATION_LENGTH = 25

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

81 """

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

83 """

84 if event.owner_user:

85 return event.owner_user_id == user_id

86 # otherwise owned by a cluster

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

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

91 """

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

93 """

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

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

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

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

100 return True

101

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

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

104

105

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

107 return (

108 _is_event_owner(event, user_id)

109 or _is_event_organizer(event, user_id)

110 or _can_moderate_event(session, event, user_id)

111 )

112

113

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

115 event = occurrence.event

116

117 next_occurrence = (

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

119 .order_by(EventOccurrence.end_time.asc())

120 .limit(1)

121 .one_or_none()

122 )

123

124 owner_community_id = None

125 owner_group_id = None

126 if event.owner_cluster:

127 if event.owner_cluster.is_official_cluster:

128 owner_community_id = event.owner_cluster.parent_node_id

129 else:

130 owner_group_id = event.owner_cluster.id

131

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

133 attendance_state = attendance.attendee_status if attendance else None

134

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

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

137

138 going_count = session.execute(

139 where_users_column_visible(

140 select(func.count())

141 .select_from(EventOccurrenceAttendee)

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

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

144 context,

145 EventOccurrenceAttendee.user_id,

146 )

147 ).scalar_one()

148 organizer_count = session.execute(

149 where_users_column_visible(

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

151 context,

152 EventOrganizer.user_id,

153 )

154 ).scalar_one()

155 subscriber_count = session.execute(

156 where_users_column_visible(

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

158 context,

159 EventSubscription.user_id,

160 )

161 ).scalar_one()

162

163 return events_pb2.Event(

164 event_id=occurrence.id,

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

166 is_cancelled=occurrence.is_cancelled,

167 is_deleted=occurrence.is_deleted,

168 title=event.title,

169 slug=event.slug,

170 content=occurrence.content,

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

172 photo_key=occurrence.photo_key or "",

173 location=events_pb2.EventLocation(

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

175 ),

176 created=Timestamp_from_datetime(occurrence.created),

177 last_edited=Timestamp_from_datetime(occurrence.last_edited),

178 creator_user_id=occurrence.creator_user_id,

179 start_time=Timestamp_from_datetime(occurrence.start_time),

180 end_time=Timestamp_from_datetime(occurrence.end_time),

181 timezone=occurrence.timezone,

182 attendance_state=attendancestate2api[attendance_state],

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

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

185 going_count=going_count,

186 organizer_count=organizer_count,

187 subscriber_count=subscriber_count,

188 owner_user_id=event.owner_user_id,

189 owner_community_id=owner_community_id,

190 owner_group_id=owner_group_id,

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

192 can_edit=can_edit,

193 can_moderate=can_moderate,

194 )

195

196

197def _get_event_and_occurrence_query(

198 occurrence_id: int,

199 include_deleted: bool,

200 context: CouchersContext | None = None,

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

202 query = (

203 select(Event, EventOccurrence)

204 .where(EventOccurrence.id == occurrence_id)

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

206 )

207

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

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

210

211 if context is not None:

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

213

214 return query

215

216

217def _get_event_and_occurrence_one(

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

219) -> tuple[Event, EventOccurrence]:

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

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

222 return result._tuple()

223

224

225def _get_event_and_occurrence_one_or_none(

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

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

228 result = session.execute(

229 _get_event_and_occurrence_query(occurrence_id, include_deleted, context=context)

230 ).one_or_none()

231 return result._tuple() if result else None

232

233

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

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

236 if not location or not location.address:

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

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

239 # No events allowed on Null Island

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

241

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

243 return (geom, location.address)

244

245

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

247 timezone_id = session.execute(

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

249 ).scalar_one_or_none()

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

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

252

253 return ZoneInfo(timezone_id)

254

255

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

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

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

259

260 try:

261 naive_datetime = datetime.fromisoformat(value)

262 except ValueError:

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

264

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

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

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

268

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

270

271

272def _update_datetime(

273 new_iso8601_local: str | None,

274 new_timezone: ZoneInfo,

275 old_datetime: datetime,

276 old_timezone: ZoneInfo,

277 context: CouchersContext,

278) -> datetime:

279 if new_iso8601_local is None and new_timezone != old_timezone:

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

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

282 if new_iso8601_local is None:

283 return old_datetime # No change

284 # New effective datetime/timestamp

285 return _check_iso8601_local_datetime(new_iso8601_local, new_timezone, context)

286

287

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

289 if start_time < now():

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

291 if end_time < start_time:

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

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

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

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

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

297

298

299def apply_occurrence_pagination(

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

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

302 """

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

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

305 """

306 if not past:

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

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

309 )

310 if page_token:

311 start_time, occurrence_id = dt_id_from_page_token(page_token)

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

313 else:

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

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

316 )

317 if page_token:

318 start_time, occurrence_id = dt_id_from_page_token(page_token)

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

320 return query

321

322

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

324 if len(occurrences) <= page_size:

325 return None

326 next_occurrence = occurrences[page_size]

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

328

329

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

331 """

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

333 """

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

335 not_already_involved = User.id.not_in(

336 select(EventOccurrenceAttendee.user_id)

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

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

339 )

340

341 cluster = occurrence.event.parent_node.official_cluster

342 creator = aliased(User)

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

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

345 return [], occurrence.event.parent_node_id

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

347 members = (

348 session.execute(

349 select(User)

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

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

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

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

354 .where(not_already_involved)

355 )

356 .scalars()

357 .all()

358 )

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

360 else:

361 max_radius = 20000 # m

362 users = (

363 session.execute(

364 select(User)

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

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

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

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

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

370 .where(not_already_involved)

371 )

372 .scalars()

373 .all()

374 )

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

376

377

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

379 """

380 Background job to generated/fan out event notifications

381 """

382 # Import here to avoid circular dependency

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

384

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

386

387 with session_scope() as session:

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

389

390 users, node_id = get_users_to_notify_for_new_event(session, occurrence)

391

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

393

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

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

396 return

397

398 for user in users:

399 context = make_notification_user_context(user_id=user.id)

400 topic_action = (

401 NotificationTopicAction.event__create_approved

402 if payload.approved

403 else NotificationTopicAction.event__create_any

404 )

405 notify(

406 session,

407 user_id=user.id,

408 topic_action=topic_action,

409 key=str(payload.occurrence_id),

410 data=notification_data_pb2.EventCreate(

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

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

413 nearby=True if node_id is None else None,

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

415 ),

416 moderation_state_id=occurrence.moderation_state_id,

417 )

418

419

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

421 with session_scope() as session:

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

423

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

425

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

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

428

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

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

431 continue

432 context = make_notification_user_context(user_id=user_id)

433 notify(

434 session,

435 user_id=user_id,

436 topic_action=NotificationTopicAction.event__update,

437 key=str(payload.occurrence_id),

438 data=notification_data_pb2.EventUpdate(

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

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

441 updated_enum_items=(

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

443 ),

444 ),

445 moderation_state_id=occurrence.moderation_state_id,

446 )

447

448

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

450 with session_scope() as session:

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

452

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

454

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

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

457

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

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

460 continue

461 context = make_notification_user_context(user_id=user_id)

462 notify(

463 session,

464 user_id=user_id,

465 topic_action=NotificationTopicAction.event__cancel,

466 key=str(payload.occurrence_id),

467 data=notification_data_pb2.EventCancel(

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

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

470 ),

471 moderation_state_id=occurrence.moderation_state_id,

472 )

473

474

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

476 with session_scope() as session:

477 event, occurrence = _get_event_and_occurrence_one(

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

479 )

480

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

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

483

484 for user_id in set(subscribed_user_ids + attending_user_ids):

485 context = make_notification_user_context(user_id=user_id)

486 notify(

487 session,

488 user_id=user_id,

489 topic_action=NotificationTopicAction.event__delete,

490 key=str(payload.occurrence_id),

491 data=notification_data_pb2.EventDelete(

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

493 ),

494 moderation_state_id=occurrence.moderation_state_id,

495 )

496

497

498class Events(events_pb2_grpc.EventsServicer):

499 def CreateEvent(

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

501 ) -> events_pb2.Event:

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

503 if not has_completed_profile(session, user):

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

505 if not request.title:

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

507 if not request.content:

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

509

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

511 timezone = _check_timezone_at(geom, context, session)

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

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

514 _check_occurrence_time_validity(start_datetime, end_datetime, context)

515

516 if request.parent_community_id:

517 parent_node = session.execute(

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

519 ).scalar_one_or_none()

520

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

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

523 else:

524 # parent community computed from geom

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

526

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

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

529

530 if (

531 request.photo_key

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

533 ):

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

535

536 thread = Thread()

537 session.add(thread)

538 session.flush()

539

540 event = Event(

541 title=request.title,

542 parent_node_id=parent_node.id,

543 owner_user_id=context.user_id,

544 thread_id=thread.id,

545 creator_user_id=context.user_id,

546 )

547 session.add(event)

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 )

565 session.add(occurrence)

566 session.flush()

567 return occurrence.id

568

569 create_moderation(

570 session=session,

571 object_type=ModerationObjectType.event_occurrence,

572 object_id=create_occurrence,

573 creator_user_id=context.user_id,

574 )

575

576 assert occurrence is not None

577

578 session.add(

579 EventOrganizer(

580 user_id=context.user_id,

581 event_id=event.id,

582 )

583 )

584

585 session.add(

586 EventSubscription(

587 user_id=context.user_id,

588 event_id=event.id,

589 )

590 )

591

592 session.add(

593 EventOccurrenceAttendee(

594 user_id=context.user_id,

595 occurrence_id=occurrence.id,

596 attendee_status=AttendeeStatus.going,

597 )

598 )

599

600 session.commit()

601

602 log_event(

603 context,

604 session,

605 "event.created",

606 {

607 "event_id": event.id,

608 "occurrence_id": occurrence.id,

609 "parent_community_id": parent_node.id,

610 "parent_community_name": parent_node.official_cluster.name,

611 },

612 )

613

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

615 queue_job(

616 session,

617 job=generate_event_create_notifications,

618 payload=jobs_pb2.GenerateEventCreateNotificationsPayload(

619 inviting_user_id=user.id,

620 occurrence_id=occurrence.id,

621 approved=False,

622 ),

623 )

624

625 return event_to_pb(session, occurrence, context)

626

627 def ScheduleEvent(

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

629 ) -> events_pb2.Event:

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

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

632

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

634 timezone = _check_timezone_at(geom, context, session)

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

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

637 _check_occurrence_time_validity(start_datetime, end_datetime, context)

638

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

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

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

642

643 event, occurrence = res

644

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

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

647

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

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

650

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

652 request.photo_key

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

654 ):

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

656

657 during = TimestamptzRange(start_datetime, end_datetime)

658

659 # && is the overlap operator for ranges

660 if (

661 session.execute(

662 select(EventOccurrence.id)

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

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

665 .limit(1)

666 )

667 .scalars()

668 .one_or_none()

669 is not None

670 ):

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

672

673 new_occurrence: EventOccurrence | None = None

674

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

676 nonlocal new_occurrence

677 new_occurrence = EventOccurrence(

678 event_id=event.id,

679 content=request.content,

680 geom=geom,

681 address=address,

682 timezone=timezone.key,

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

684 during=during,

685 creator_user_id=context.user_id,

686 moderation_state_id=moderation_state_id,

687 )

688 session.add(new_occurrence)

689 session.flush()

690 return new_occurrence.id

691

692 create_moderation(

693 session=session,

694 object_type=ModerationObjectType.event_occurrence,

695 object_id=create_occurrence,

696 creator_user_id=context.user_id,

697 )

698

699 assert new_occurrence is not None

700

701 session.add(

702 EventOccurrenceAttendee(

703 user_id=context.user_id,

704 occurrence_id=new_occurrence.id,

705 attendee_status=AttendeeStatus.going,

706 )

707 )

708

709 session.flush()

710

711 # TODO: notify

712

713 return event_to_pb(session, new_occurrence, context)

714

715 def UpdateEvent(

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

717 ) -> events_pb2.Event:

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

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

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

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

722

723 event, occurrence = res

724

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

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

727

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

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

730

731 if occurrence.is_cancelled:

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

733

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

735

736 if request.HasField("title"):

737 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_TITLE)

738 event.title = request.title.value

739

740 if request.HasField("content"):

741 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_CONTENT)

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

743

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

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

746

747 old_timezone = ZoneInfo(occurrence.timezone)

748 timezone: ZoneInfo = old_timezone

749 if request.HasField("location"):

750 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_LOCATION)

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

752 timezone = _check_timezone_at(geom, context, session)

753 occurrence_update["geom"] = geom

754 occurrence_update["address"] = address

755 occurrence_update["timezone"] = timezone.key

756

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

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

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

760

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

762 start_datetime = _update_datetime(

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

764 timezone,

765 old_datetime=occurrence.start_time,

766 old_timezone=old_timezone,

767 context=context,

768 )

769 if start_datetime != occurrence.start_time:

770 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_START_TIME)

771

772 end_datetime = _update_datetime(

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

774 timezone,

775 old_datetime=occurrence.end_time,

776 old_timezone=old_timezone,

777 context=context,

778 )

779 if end_datetime != occurrence.end_time:

780 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_END_TIME)

781

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

783 _check_occurrence_time_validity(start_datetime, end_datetime, context)

784

785 during = TimestamptzRange(start_datetime, end_datetime)

786

787 # && is the overlap operator for ranges

788 if (

789 session.execute(

790 select(EventOccurrence.id)

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

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

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

794 .limit(1)

795 )

796 .scalars()

797 .one_or_none()

798 is not None

799 ):

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

801

802 occurrence_update["during"] = during

803

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

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

806

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

808 if request.update_all_future:

809 session.execute(

810 update(EventOccurrence)

811 .where(EventOccurrence.end_time >= cutoff_time)

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

813 .values(occurrence_update)

814 .execution_options(synchronize_session=False)

815 )

816 else:

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

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

819 session.execute(

820 update(EventOccurrence)

821 .where(EventOccurrence.end_time >= cutoff_time)

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

823 .values(occurrence_update)

824 .execution_options(synchronize_session=False)

825 )

826

827 session.flush()

828

829 if notify_updated:

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

831 if request.should_notify:

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

833

834 queue_job(

835 session,

836 job=generate_event_update_notifications,

837 payload=jobs_pb2.GenerateEventUpdateNotificationsPayload(

838 updating_user_id=user.id,

839 occurrence_id=occurrence.id,

840 updated_enum_items=notify_updated,

841 ),

842 )

843 else:

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

845

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

847 session.refresh(occurrence)

848

849 return event_to_pb(session, occurrence, context)

850

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

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

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

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

855

856 if not occurrence:

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

858

859 return event_to_pb(session, occurrence, context)

860

861 def CancelEvent(

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

863 ) -> empty_pb2.Empty:

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

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

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

867

868 event, occurrence = res

869

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

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

872

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

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

875

876 occurrence.is_cancelled = True

877

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

879

880 queue_job(

881 session,

882 job=generate_event_cancel_notifications,

883 payload=jobs_pb2.GenerateEventCancelNotificationsPayload(

884 cancelling_user_id=context.user_id,

885 occurrence_id=occurrence.id,

886 ),

887 )

888

889 return empty_pb2.Empty()

890

891 def RequestCommunityInvite(

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

893 ) -> empty_pb2.Empty:

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

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

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

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

898

899 event, occurrence = res

900

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

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

903

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

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

906

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

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

909

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

911

912 if len(this_user_reqs) > 0:

913 context.abort_with_error_code(

914 grpc.StatusCode.FAILED_PRECONDITION, "event_community_invite_already_requested"

915 )

916

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

918

919 if len(approved_reqs) > 0:

920 context.abort_with_error_code(

921 grpc.StatusCode.FAILED_PRECONDITION, "event_community_invite_already_approved"

922 )

923

924 req = EventCommunityInviteRequest(

925 occurrence_id=request.event_id,

926 user_id=context.user_id,

927 )

928 session.add(req)

929 session.flush()

930

931 send_event_community_invite_request_email(session, req)

932

933 return empty_pb2.Empty()

934

935 def ListEventOccurrences(

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

937 ) -> events_pb2.ListEventOccurrencesRes:

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

939 initial_query = (

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

941 )

942 initial_query = where_moderated_content_visible(

943 initial_query, context, EventOccurrence, is_list_operation=False

944 )

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

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

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

948

949 query = (

950 select(EventOccurrence)

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

952 .where(~EventOccurrence.is_deleted)

953 )

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

955

956 if not request.include_cancelled:

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

958

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

960

961 query = query.limit(page_size + 1)

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

963

964 return events_pb2.ListEventOccurrencesRes(

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

966 next_page_token=occurrences_next_page_token(occurrences, page_size),

967 )

968

969 def ListEventAttendees(

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

971 ) -> events_pb2.ListEventAttendeesRes:

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

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

974 occurrence = session.execute(

975 where_moderated_content_visible(

976 select(EventOccurrence)

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

978 .where(~EventOccurrence.is_deleted),

979 context,

980 EventOccurrence,

981 is_list_operation=False,

982 )

983 ).scalar_one_or_none()

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

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

986 attendees = (

987 session.execute(

988 where_users_column_visible(

989 select(EventOccurrenceAttendee)

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

991 .where(EventOccurrenceAttendee.user_id >= next_user_id)

992 .order_by(EventOccurrenceAttendee.user_id)

993 .limit(page_size + 1),

994 context,

995 EventOccurrenceAttendee.user_id,

996 )

997 )

998 .scalars()

999 .all()

1000 )

1001 return events_pb2.ListEventAttendeesRes(

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

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

1004 )

1005

1006 def ListEventSubscribers(

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

1008 ) -> events_pb2.ListEventSubscribersRes:

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

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

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

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

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

1014 event, occurrence = res

1015 subscribers = (

1016 session.execute(

1017 where_users_column_visible(

1018 select(EventSubscription)

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

1020 .where(EventSubscription.user_id >= next_user_id)

1021 .order_by(EventSubscription.user_id)

1022 .limit(page_size + 1),

1023 context,

1024 EventSubscription.user_id,

1025 )

1026 )

1027 .scalars()

1028 .all()

1029 )

1030 return events_pb2.ListEventSubscribersRes(

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

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

1033 )

1034

1035 def ListEventOrganizers(

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

1037 ) -> events_pb2.ListEventOrganizersRes:

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

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

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

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

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

1043 event, occurrence = res

1044 organizers = (

1045 session.execute(

1046 where_users_column_visible(

1047 select(EventOrganizer)

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

1049 .where(EventOrganizer.user_id >= next_user_id)

1050 .order_by(EventOrganizer.user_id)

1051 .limit(page_size + 1),

1052 context,

1053 EventOrganizer.user_id,

1054 )

1055 )

1056 .scalars()

1057 .all()

1058 )

1059 return events_pb2.ListEventOrganizersRes(

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

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

1062 )

1063

1064 def TransferEvent(

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

1066 ) -> events_pb2.Event:

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

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

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

1070

1071 event, occurrence = res

1072

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

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

1075

1076 if occurrence.is_cancelled:

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

1078

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

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

1081

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

1083 cluster = session.execute(

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

1085 ).scalar_one_or_none()

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

1087 cluster = session.execute(

1088 select(Cluster)

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

1090 .where(Cluster.is_official_cluster)

1091 ).scalar_one_or_none()

1092

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

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

1095

1096 event.owner_user = None

1097 event.owner_cluster = cluster

1098

1099 session.commit()

1100 return event_to_pb(session, occurrence, context)

1101

1102 def SetEventSubscription(

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

1104 ) -> events_pb2.Event:

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

1109 event, occurrence = res

1110

1111 if occurrence.is_cancelled:

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

1113

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

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

1116

1117 current_subscription = session.execute(

1118 select(EventSubscription)

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

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

1121 ).scalar_one_or_none()

1122

1123 # if not subscribed, subscribe

1124 if request.subscribe and not current_subscription:

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

1126

1127 # if subscribed but unsubbing, remove subscription

1128 if not request.subscribe and current_subscription:

1129 session.delete(current_subscription)

1130

1131 session.flush()

1132

1133 log_event(

1134 context,

1135 session,

1136 "event.subscription_set",

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

1138 )

1139

1140 return event_to_pb(session, occurrence, context)

1141

1142 def SetEventAttendance(

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

1144 ) -> events_pb2.Event:

1145 occurrence = session.execute(

1146 where_moderated_content_visible(

1147 select(EventOccurrence)

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

1149 .where(~EventOccurrence.is_deleted),

1150 context,

1151 EventOccurrence,

1152 is_list_operation=False,

1153 )

1154 ).scalar_one_or_none()

1155

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

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

1158

1159 if occurrence.is_cancelled:

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

1161

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

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

1164

1165 current_attendance = session.execute(

1166 select(EventOccurrenceAttendee)

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

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

1169 ).scalar_one_or_none()

1170

1171 if request.attendance_state == events_pb2.ATTENDANCE_STATE_NOT_GOING:

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

1173 session.delete(current_attendance)

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

1175 else:

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

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

1178 else:

1179 # create new

1180 attendance = EventOccurrenceAttendee(

1181 user_id=context.user_id,

1182 occurrence_id=occurrence.id,

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

1184 )

1185 session.add(attendance)

1186

1187 session.flush()

1188

1189 log_event(

1190 context,

1191 session,

1192 "event.attendance_set",

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

1194 )

1195

1196 return event_to_pb(session, occurrence, context)

1197

1198 def ListMyEvents(

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

1200 ) -> events_pb2.ListMyEventsRes:

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

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

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

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

1205 page_number = request.page_number or 1

1206 # Calculate the offset for pagination

1207 offset = (page_number - 1) * page_size

1208 query = (

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

1210 )

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

1212

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

1214 include_subscribed = request.subscribed or include_all

1215 include_organizing = request.organizing or include_all

1216 include_attending = request.attending or include_all

1217 include_my_communities = request.my_communities or include_all

1218

1219 if include_attending and request.exclude_attending:

1220 context.abort_with_error_code(

1221 grpc.StatusCode.INVALID_ARGUMENT, "cannot_combine_attending_and_exclude_attending"

1222 )

1223

1224 where_ = []

1225

1226 if include_subscribed:

1227 query = query.outerjoin(

1228 EventSubscription,

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

1230 )

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

1232 if include_organizing:

1233 query = query.outerjoin(

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

1235 )

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

1237 if include_attending or request.exclude_attending:

1238 query = query.outerjoin(

1239 EventOccurrenceAttendee,

1240 and_(

1241 EventOccurrenceAttendee.occurrence_id == EventOccurrence.id,

1242 EventOccurrenceAttendee.user_id == context.user_id,

1243 ),

1244 )

1245 if include_attending:

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

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

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

1249 query = query.outerjoin(

1250 EventOrganizer,

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

1252 )

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

1254 if include_my_communities:

1255 my_communities = (

1256 session.execute(

1257 select(Node.id)

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

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

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

1261 .where(Cluster.is_official_cluster)

1262 .order_by(Node.id)

1263 .limit(100000)

1264 )

1265 .scalars()

1266 .all()

1267 )

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

1269

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

1271

1272 if request.my_communities_exclude_global:

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

1274

1275 if not request.include_cancelled:

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

1277

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

1279 # Count the total number of items for pagination

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

1281 # Apply pagination by page number

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

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

1284

1285 return events_pb2.ListMyEventsRes(

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

1287 next_page_token=occurrences_next_page_token(occurrences, page_size),

1288 total_items=total_items,

1289 )

1290

1291 def ListAllEvents(

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

1293 ) -> events_pb2.ListAllEventsRes:

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

1295

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

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

1298

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

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

1301

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

1303

1304 query = query.limit(page_size + 1)

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

1306

1307 return events_pb2.ListAllEventsRes(

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

1309 next_page_token=occurrences_next_page_token(occurrences, page_size),

1310 )

1311

1312 def InviteEventOrganizer(

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

1314 ) -> empty_pb2.Empty:

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

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

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

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

1319

1320 event, occurrence = res

1321

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

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

1324

1325 if occurrence.is_cancelled:

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

1327

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

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

1330

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

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

1333 ).scalar_one_or_none():

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

1335

1336 session.add(

1337 EventOrganizer(

1338 user_id=request.user_id,

1339 event_id=event.id,

1340 )

1341 )

1342 session.flush()

1343

1344 other_user_context = make_notification_user_context(user_id=request.user_id)

1345

1346 notify(

1347 session,

1348 user_id=request.user_id,

1349 topic_action=NotificationTopicAction.event__invite_organizer,

1350 key=str(event.id),

1351 data=notification_data_pb2.EventInviteOrganizer(

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

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

1354 ),

1355 )

1356

1357 return empty_pb2.Empty()

1358

1359 def RemoveEventOrganizer(

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

1361 ) -> empty_pb2.Empty:

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

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

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

1365

1366 event, occurrence = res

1367

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

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

1370

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

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

1373

1374 # Determine which user to remove

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

1376

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

1378 if event.owner_user_id == user_id_to_remove:

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

1380

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

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

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

1384

1385 # Find the organizer to remove

1386 organizer_to_remove = session.execute(

1387 select(EventOrganizer)

1388 .where(EventOrganizer.user_id == user_id_to_remove)

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

1390 ).scalar_one_or_none()

1391

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

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

1394

1395 session.delete(organizer_to_remove)

1396

1397 return empty_pb2.Empty()

1398

1399 def GetEventCalendarFile(

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

1401 ) -> httpbody_pb2.HttpBody:

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

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

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

1405

1406 _, occurrence_db = res

1407

1408 event_pb = event_to_pb(session, occurrence_db, context)

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

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