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

1import logging

2from datetime import datetime, timedelta

3from typing import Any, cast

5import grpc

6from google.protobuf import empty_pb2

7from psycopg.types.range import TimestamptzRange

8from sqlalchemy import Select, select

9from sqlalchemy.orm import Session

10from sqlalchemy.sql import and_, func, or_, update

12from couchers.context import CouchersContext, make_notification_user_context

13from couchers.db import can_moderate_node, get_parent_node_at_location, session_scope

14from couchers.event_log import log_event

15from couchers.helpers.completed_profile import has_completed_profile

16from couchers.jobs.enqueue import queue_job

17from couchers.models import (

18 AttendeeStatus,

19 Cluster,

20 ClusterSubscription,

21 Event,

22 EventCommunityInviteRequest,

23 EventOccurrence,

24 EventOccurrenceAttendee,

25 EventOrganizer,

26 EventSubscription,

27 ModerationObjectType,

28 Node,

29 NodeType,

30 Thread,

31 Upload,

32 User,

33)

34from couchers.models.notifications import NotificationTopicAction

35from couchers.moderation.utils import create_moderation

36from couchers.notifications.notify import notify

37from couchers.proto import events_pb2, events_pb2_grpc, notification_data_pb2

38from couchers.proto.internal import jobs_pb2

39from couchers.servicers.api import user_model_to_pb

40from couchers.servicers.blocking import is_not_visible

41from couchers.servicers.threads import thread_to_pb

42from couchers.sql import users_visible, where_moderated_content_visible, where_users_column_visible

43from couchers.tasks import send_event_community_invite_request_email

44from couchers.utils import (

45 Timestamp_from_datetime,

46 create_coordinate,

47 dt_from_millis,

48 millis_from_dt,

49 not_none,

50 now,

51 to_aware_datetime,

52)

54logger = logging.getLogger(__name__)

56attendancestate2sql = {

57 events_pb2.AttendanceState.ATTENDANCE_STATE_NOT_GOING: None,

58 events_pb2.AttendanceState.ATTENDANCE_STATE_GOING: AttendeeStatus.going,

59}

61attendancestate2api = {

62 None: events_pb2.AttendanceState.ATTENDANCE_STATE_NOT_GOING,

63 AttendeeStatus.going: events_pb2.AttendanceState.ATTENDANCE_STATE_GOING,

64}

66MAX_PAGINATION_LENGTH = 25

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

70 """

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

72 """

73 if event.owner_user:

74 return event.owner_user_id == user_id

75 # otherwise owned by a cluster

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

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

80 """

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

82 """

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

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

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

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

89 return True

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

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

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

96 return (

97 _is_event_owner(event, user_id)

98 or _is_event_organizer(event, user_id)

99 or _can_moderate_event(session, event, user_id)

100 )

101

102

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

104 event = occurrence.event

105

106 next_occurrence = (

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

108 .order_by(EventOccurrence.end_time.asc())

109 .limit(1)

110 .one_or_none()

111 )

112

113 owner_community_id = None

114 owner_group_id = None

115 if event.owner_cluster:

116 if event.owner_cluster.is_official_cluster:

117 owner_community_id = event.owner_cluster.parent_node_id

118 else:

119 owner_group_id = event.owner_cluster.id

120

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

122 attendance_state = attendance.attendee_status if attendance else None

123

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

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

126

127 going_count = session.execute(

128 where_users_column_visible(

129 select(func.count())

130 .select_from(EventOccurrenceAttendee)

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

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

133 context,

134 EventOccurrenceAttendee.user_id,

135 )

136 ).scalar_one()

137 organizer_count = session.execute(

138 where_users_column_visible(

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

140 context,

141 EventOrganizer.user_id,

142 )

143 ).scalar_one()

144 subscriber_count = session.execute(

145 where_users_column_visible(

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

147 context,

148 EventSubscription.user_id,

149 )

150 ).scalar_one()

151

152 location: events_pb2.EventLocation

153 if occurrence.geom:

154 location = events_pb2.EventLocation(

155 lat=not_none(occurrence.coordinates)[0], lng=not_none(occurrence.coordinates)[1], address=occurrence.address

156 )

157 else:

158 # Backcompat: Surface legacy online events as offline events.

159 # They'll appear at null island, but there are so few we're ok with this.

160 location = events_pb2.EventLocation(address=occurrence.link, lat=0, lng=0)

161

162 return events_pb2.Event(

163 event_id=occurrence.id,

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

165 is_cancelled=occurrence.is_cancelled,

166 is_deleted=occurrence.is_deleted,

167 title=event.title,

168 slug=event.slug,

169 content=occurrence.content,

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

171 photo_key=occurrence.photo_key or "",

172 location=location,

173 created=Timestamp_from_datetime(occurrence.created),

174 last_edited=Timestamp_from_datetime(occurrence.last_edited),

175 creator_user_id=occurrence.creator_user_id,

176 start_time=Timestamp_from_datetime(occurrence.start_time),

177 end_time=Timestamp_from_datetime(occurrence.end_time),

178 timezone=occurrence.timezone,

179 attendance_state=attendancestate2api[attendance_state],

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

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

182 going_count=going_count,

183 organizer_count=organizer_count,

184 subscriber_count=subscriber_count,

185 owner_user_id=event.owner_user_id,

186 owner_community_id=owner_community_id,

187 owner_group_id=owner_group_id,

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

189 can_edit=can_edit,

190 can_moderate=can_moderate,

191 )

192

193

194def _get_event_and_occurrence_query(

195 occurrence_id: int,

196 include_deleted: bool,

197 context: CouchersContext | None = None,

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

199 query = (

200 select(Event, EventOccurrence)

201 .where(EventOccurrence.id == occurrence_id)

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

203 )

204

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

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

207

208 if context is not None:

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

210

211 return query

212

213

214def _get_event_and_occurrence_one(

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

216) -> tuple[Event, EventOccurrence]:

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

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

219 return result._tuple()

220

221

222def _get_event_and_occurrence_one_or_none(

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

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

225 result = session.execute(

226 _get_event_and_occurrence_query(occurrence_id, include_deleted, context=context)

227 ).one_or_none()

228 return result._tuple() if result else None

229

230

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

232 if start_time < now():

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

234 if end_time < start_time:

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

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

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

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

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

240

241

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

243 """

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

245 """

246 cluster = occurrence.event.parent_node.official_cluster

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

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

249 return [], occurrence.event.parent_node_id

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

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

252 else:

253 max_radius = 20000 # m

254 users = (

255 session.execute(

256 select(User)

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

258 .where(User.is_visible)

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

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

261 )

262 .scalars()

263 .all()

264 )

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

266

267

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

269 """

270 Background job to generated/fan out event notifications

271 """

272 # Import here to avoid circular dependency

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

274

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

276

277 with session_scope() as session:

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

279 creator = occurrence.creator_user

280

281 users, node_id = get_users_to_notify_for_new_event(session, occurrence)

282

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

284

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

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

287 return

288

289 for user in users:

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

291 continue

292 context = make_notification_user_context(user_id=user.id)

293 topic_action = (

294 NotificationTopicAction.event__create_approved

295 if payload.approved

296 else NotificationTopicAction.event__create_any

297 )

298 notify(

299 session,

300 user_id=user.id,

301 topic_action=topic_action,

302 key=str(payload.occurrence_id),

303 data=notification_data_pb2.EventCreate(

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

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

306 nearby=True if node_id is None else None,

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

308 ),

309 moderation_state_id=occurrence.moderation_state_id,

310 )

311

312

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

314 with session_scope() as session:

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

316

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

318

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

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

321

322 for user_id in set(subscribed_user_ids + attending_user_ids):

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

324 continue

325 context = make_notification_user_context(user_id=user_id)

326 notify(

327 session,

328 user_id=user_id,

329 topic_action=NotificationTopicAction.event__update,

330 key=str(payload.occurrence_id),

331 data=notification_data_pb2.EventUpdate(

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

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

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

335 updated_str_items=payload.updated_str_items,

336 updated_enum_items=(

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

338 ),

339 ),

340 moderation_state_id=occurrence.moderation_state_id,

341 )

342

343

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

345 with session_scope() as session:

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

347

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

349

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

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

352

353 for user_id in set(subscribed_user_ids + attending_user_ids):

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

355 continue

356 context = make_notification_user_context(user_id=user_id)

357 notify(

358 session,

359 user_id=user_id,

360 topic_action=NotificationTopicAction.event__cancel,

361 key=str(payload.occurrence_id),

362 data=notification_data_pb2.EventCancel(

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

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

365 ),

366 moderation_state_id=occurrence.moderation_state_id,

367 )

368

369

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

371 with session_scope() as session:

372 event, occurrence = _get_event_and_occurrence_one(

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

374 )

375

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

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

378

379 for user_id in set(subscribed_user_ids + attending_user_ids):

380 context = make_notification_user_context(user_id=user_id)

381 notify(

382 session,

383 user_id=user_id,

384 topic_action=NotificationTopicAction.event__delete,

385 key=str(payload.occurrence_id),

386 data=notification_data_pb2.EventDelete(

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

388 ),

389 moderation_state_id=occurrence.moderation_state_id,

390 )

391

392

393class Events(events_pb2_grpc.EventsServicer):

394 def CreateEvent(

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

396 ) -> events_pb2.Event:

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

398 if not has_completed_profile(session, user):

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

400 if not request.title:

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

402 if not request.content:

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

404

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

406 if not (

407 request.HasField("location") and request.location.address and request.location.lat and request.location.lng

408 ):

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

410 if request.location.lat == 0 and request.location.lng == 0: 410 ↛ 411line 410 didn't jump to line 411 because the condition on line 410 was never true

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

412 geom = create_coordinate(request.location.lat, request.location.lng)

413 address = request.location.address

414

415 start_time = to_aware_datetime(request.start_time)

416 end_time = to_aware_datetime(request.end_time)

417

418 _check_occurrence_time_validity(start_time, end_time, context)

419

420 if request.parent_community_id:

421 parent_node = session.execute(

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

423 ).scalar_one_or_none()

424

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

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

427 else:

428 # parent community computed from geom

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

430

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

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

433

434 if (

435 request.photo_key

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

437 ):

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

439

440 thread = Thread()

441 session.add(thread)

442 session.flush()

443

444 event = Event(

445 title=request.title,

446 parent_node_id=parent_node.id,

447 owner_user_id=context.user_id,

448 thread_id=thread.id,

449 creator_user_id=context.user_id,

450 )

451 session.add(event)

452 session.flush()

453

454 occurrence: EventOccurrence | None = None

455

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

457 nonlocal occurrence

458 occurrence = EventOccurrence(

459 event_id=event.id,

460 content=request.content,

461 geom=geom,

462 address=address,

463 link=None,

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

465 # timezone=timezone,

466 during=TimestamptzRange(start_time, end_time),

467 creator_user_id=context.user_id,

468 moderation_state_id=moderation_state_id,

469 )

470 session.add(occurrence)

471 session.flush()

472 return occurrence.id

473

474 create_moderation(

475 session=session,

476 object_type=ModerationObjectType.event_occurrence,

477 object_id=create_occurrence,

478 creator_user_id=context.user_id,

479 )

480

481 assert occurrence is not None

482

483 session.add(

484 EventOrganizer(

485 user_id=context.user_id,

486 event_id=event.id,

487 )

488 )

489

490 session.add(

491 EventSubscription(

492 user_id=context.user_id,

493 event_id=event.id,

494 )

495 )

496

497 session.add(

498 EventOccurrenceAttendee(

499 user_id=context.user_id,

500 occurrence_id=occurrence.id,

501 attendee_status=AttendeeStatus.going,

502 )

503 )

504

505 session.commit()

506

507 log_event(

508 context,

509 session,

510 "event.created",

511 {

512 "event_id": event.id,

513 "occurrence_id": occurrence.id,

514 "parent_community_id": parent_node.id,

515 "parent_community_name": parent_node.official_cluster.name,

516 },

517 )

518

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

520 queue_job(

521 session,

522 job=generate_event_create_notifications,

523 payload=jobs_pb2.GenerateEventCreateNotificationsPayload(

524 inviting_user_id=user.id,

525 occurrence_id=occurrence.id,

526 approved=False,

527 ),

528 )

529

530 return event_to_pb(session, occurrence, context)

531

532 def ScheduleEvent(

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

534 ) -> events_pb2.Event:

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

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

537 if not ( 537 ↛ 540line 537 didn't jump to line 540 because the condition on line 537 was never true

538 request.HasField("location") and request.location.address and request.location.lat and request.location.lng

539 ):

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

541 if request.location.lat == 0 and request.location.lng == 0: 541 ↛ 542line 541 didn't jump to line 542 because the condition on line 541 was never true

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

543 geom = create_coordinate(request.location.lat, request.location.lng)

544 address = request.location.address

545

546 start_time = to_aware_datetime(request.start_time)

547 end_time = to_aware_datetime(request.end_time)

548

549 _check_occurrence_time_validity(start_time, end_time, context)

550

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

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

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

554

555 event, occurrence = res

556

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

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

559

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

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

562

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

564 request.photo_key

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

566 ):

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

568

569 during = TimestamptzRange(start_time, end_time)

570

571 # && is the overlap operator for ranges

572 if (

573 session.execute(

574 select(EventOccurrence.id)

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

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

577 .limit(1)

578 )

579 .scalars()

580 .one_or_none()

581 is not None

582 ):

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

584

585 new_occurrence: EventOccurrence | None = None

586

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

588 nonlocal new_occurrence

589 new_occurrence = EventOccurrence(

590 event_id=event.id,

591 content=request.content,

592 geom=geom,

593 address=address,

594 link=None,

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

596 # timezone=timezone,

597 during=during,

598 creator_user_id=context.user_id,

599 moderation_state_id=moderation_state_id,

600 )

601 session.add(new_occurrence)

602 session.flush()

603 return new_occurrence.id

604

605 create_moderation(

606 session=session,

607 object_type=ModerationObjectType.event_occurrence,

608 object_id=create_occurrence,

609 creator_user_id=context.user_id,

610 )

611

612 assert new_occurrence is not None

613

614 session.add(

615 EventOccurrenceAttendee(

616 user_id=context.user_id,

617 occurrence_id=new_occurrence.id,

618 attendee_status=AttendeeStatus.going,

619 )

620 )

621

622 session.flush()

623

624 # TODO: notify

625

626 return event_to_pb(session, new_occurrence, context)

627

628 def UpdateEvent(

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

630 ) -> events_pb2.Event:

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

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

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

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

635

636 event, occurrence = res

637

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

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

640

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

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

643

644 if occurrence.is_cancelled:

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

646

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

648

649 if request.HasField("title"):

650 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_TITLE)

651 event.title = request.title.value

652

653 if request.HasField("content"):

654 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_CONTENT)

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

656

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

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

659

660 if request.HasField("location"):

661 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_LOCATION)

662 occurrence_update["link"] = None

663 if request.location.lat == 0 and request.location.lng == 0: 663 ↛ 664line 663 didn't jump to line 664 because the condition on line 663 was never true

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

665 occurrence_update["geom"] = create_coordinate(request.location.lat, request.location.lng)

666 occurrence_update["address"] = request.location.address

667

668 if request.HasField("start_time") or request.HasField("end_time"):

669 if request.update_all_future: 669 ↛ 670line 669 didn't jump to line 670 because the condition on line 669 was never true

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

671 if request.HasField("start_time"): 671 ↛ 675line 671 didn't jump to line 675 because the condition on line 671 was always true

672 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_START_TIME)

673 start_time = to_aware_datetime(request.start_time)

674 else:

675 start_time = occurrence.start_time

676 if request.HasField("end_time"):

677 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_END_TIME)

678 end_time = to_aware_datetime(request.end_time)

679 else:

680 end_time = occurrence.end_time

681

682 _check_occurrence_time_validity(start_time, end_time, context)

683

684 during = TimestamptzRange(start_time, end_time)

685

686 # && is the overlap operator for ranges

687 if (

688 session.execute(

689 select(EventOccurrence.id)

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

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

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

693 .limit(1)

694 )

695 .scalars()

696 .one_or_none()

697 is not None

698 ):

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

700

701 occurrence_update["during"] = during

702

703 # TODO

704 # if request.HasField("timezone"):

705 # occurrence_update["timezone"] = request.timezone

706

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

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

709

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

711 if request.update_all_future:

712 session.execute(

713 update(EventOccurrence)

714 .where(EventOccurrence.end_time >= cutoff_time)

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

716 .values(occurrence_update)

717 .execution_options(synchronize_session=False)

718 )

719 else:

720 if occurrence.end_time < cutoff_time: 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.FAILED_PRECONDITION, "event_cant_update_old_event")

722 session.execute(

723 update(EventOccurrence)

724 .where(EventOccurrence.end_time >= cutoff_time)

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

726 .values(occurrence_update)

727 .execution_options(synchronize_session=False)

728 )

729

730 session.flush()

731

732 if notify_updated:

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

734 if request.should_notify:

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

736

737 queue_job(

738 session,

739 job=generate_event_update_notifications,

740 payload=jobs_pb2.GenerateEventUpdateNotificationsPayload(

741 updating_user_id=user.id,

742 occurrence_id=occurrence.id,

743 updated_enum_items=notify_updated,

744 ),

745 )

746 else:

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

748

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

750 session.refresh(occurrence)

751

752 return event_to_pb(session, occurrence, context)

753

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

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

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

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

758

759 if not occurrence:

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

761

762 return event_to_pb(session, occurrence, context)

763

764 def CancelEvent(

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

766 ) -> empty_pb2.Empty:

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

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

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

770

771 event, occurrence = res

772

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

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

775

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

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

778

779 occurrence.is_cancelled = True

780

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

782

783 queue_job(

784 session,

785 job=generate_event_cancel_notifications,

786 payload=jobs_pb2.GenerateEventCancelNotificationsPayload(

787 cancelling_user_id=context.user_id,

788 occurrence_id=occurrence.id,

789 ),

790 )

791

792 return empty_pb2.Empty()

793

794 def RequestCommunityInvite(

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

796 ) -> empty_pb2.Empty:

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

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

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

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

801

802 event, occurrence = res

803

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

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

806

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

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

809

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

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

812

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

814

815 if len(this_user_reqs) > 0:

816 context.abort_with_error_code(

817 grpc.StatusCode.FAILED_PRECONDITION, "event_community_invite_already_requested"

818 )

819

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

821

822 if len(approved_reqs) > 0:

823 context.abort_with_error_code(

824 grpc.StatusCode.FAILED_PRECONDITION, "event_community_invite_already_approved"

825 )

826

827 req = EventCommunityInviteRequest(

828 occurrence_id=request.event_id,

829 user_id=context.user_id,

830 )

831 session.add(req)

832 session.flush()

833

834 send_event_community_invite_request_email(session, req)

835

836 return empty_pb2.Empty()

837

838 def ListEventOccurrences(

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

840 ) -> events_pb2.ListEventOccurrencesRes:

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

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

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

844 initial_query = (

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

846 )

847 initial_query = where_moderated_content_visible(

848 initial_query, context, EventOccurrence, is_list_operation=False

849 )

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

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

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

853

854 query = (

855 select(EventOccurrence)

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

857 .where(~EventOccurrence.is_deleted)

858 )

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

860

861 if not request.include_cancelled:

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

863

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

865 cutoff = page_token - timedelta(seconds=1)

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

867 else:

868 cutoff = page_token + timedelta(seconds=1)

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

870

871 query = query.limit(page_size + 1)

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

873

874 return events_pb2.ListEventOccurrencesRes(

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

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

877 )

878

879 def ListEventAttendees(

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

881 ) -> events_pb2.ListEventAttendeesRes:

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

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

884 occurrence = session.execute(

885 where_moderated_content_visible(

886 select(EventOccurrence)

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

888 .where(~EventOccurrence.is_deleted),

889 context,

890 EventOccurrence,

891 is_list_operation=False,

892 )

893 ).scalar_one_or_none()

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

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

896 attendees = (

897 session.execute(

898 where_users_column_visible(

899 select(EventOccurrenceAttendee)

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

901 .where(EventOccurrenceAttendee.user_id >= next_user_id)

902 .order_by(EventOccurrenceAttendee.user_id)

903 .limit(page_size + 1),

904 context,

905 EventOccurrenceAttendee.user_id,

906 )

907 )

908 .scalars()

909 .all()

910 )

911 return events_pb2.ListEventAttendeesRes(

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

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

914 )

915

916 def ListEventSubscribers(

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

918 ) -> events_pb2.ListEventSubscribersRes:

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

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

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

922 if not res: 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.NOT_FOUND, "event_not_found")

924 event, occurrence = res

925 subscribers = (

926 session.execute(

927 where_users_column_visible(

928 select(EventSubscription)

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

930 .where(EventSubscription.user_id >= next_user_id)

931 .order_by(EventSubscription.user_id)

932 .limit(page_size + 1),

933 context,

934 EventSubscription.user_id,

935 )

936 )

937 .scalars()

938 .all()

939 )

940 return events_pb2.ListEventSubscribersRes(

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

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

943 )

944

945 def ListEventOrganizers(

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

947 ) -> events_pb2.ListEventOrganizersRes:

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

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

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

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

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

953 event, occurrence = res

954 organizers = (

955 session.execute(

956 where_users_column_visible(

957 select(EventOrganizer)

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

959 .where(EventOrganizer.user_id >= next_user_id)

960 .order_by(EventOrganizer.user_id)

961 .limit(page_size + 1),

962 context,

963 EventOrganizer.user_id,

964 )

965 )

966 .scalars()

967 .all()

968 )

969 return events_pb2.ListEventOrganizersRes(

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

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

972 )

973

974 def TransferEvent(

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

976 ) -> events_pb2.Event:

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

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

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

980

981 event, occurrence = res

982

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

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

985

986 if occurrence.is_cancelled:

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

988

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

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

991

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

993 cluster = session.execute(

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

995 ).scalar_one_or_none()

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

997 cluster = session.execute(

998 select(Cluster)

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

1000 .where(Cluster.is_official_cluster)

1001 ).scalar_one_or_none()

1002

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

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

1005

1006 event.owner_user = None

1007 event.owner_cluster = cluster

1008

1009 session.commit()

1010 return event_to_pb(session, occurrence, context)

1011

1012 def SetEventSubscription(

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

1014 ) -> events_pb2.Event:

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

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

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

1018

1019 event, occurrence = res

1020

1021 if occurrence.is_cancelled:

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

1023

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

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

1026

1027 current_subscription = session.execute(

1028 select(EventSubscription)

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

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

1031 ).scalar_one_or_none()

1032

1033 # if not subscribed, subscribe

1034 if request.subscribe and not current_subscription:

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

1036

1037 # if subscribed but unsubbing, remove subscription

1038 if not request.subscribe and current_subscription:

1039 session.delete(current_subscription)

1040

1041 session.flush()

1042

1043 log_event(

1044 context,

1045 session,

1046 "event.subscription_set",

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

1048 )

1049

1050 return event_to_pb(session, occurrence, context)

1051

1052 def SetEventAttendance(

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

1054 ) -> events_pb2.Event:

1055 occurrence = session.execute(

1056 where_moderated_content_visible(

1057 select(EventOccurrence)

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

1059 .where(~EventOccurrence.is_deleted),

1060 context,

1061 EventOccurrence,

1062 is_list_operation=False,

1063 )

1064 ).scalar_one_or_none()

1065

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

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

1068

1069 if occurrence.is_cancelled:

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

1071

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

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

1074

1075 current_attendance = session.execute(

1076 select(EventOccurrenceAttendee)

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

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

1079 ).scalar_one_or_none()

1080

1081 if request.attendance_state == events_pb2.ATTENDANCE_STATE_NOT_GOING:

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

1083 session.delete(current_attendance)

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

1085 else:

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

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

1088 else:

1089 # create new

1090 attendance = EventOccurrenceAttendee(

1091 user_id=context.user_id,

1092 occurrence_id=occurrence.id,

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

1094 )

1095 session.add(attendance)

1096

1097 session.flush()

1098

1099 log_event(

1100 context,

1101 session,

1102 "event.attendance_set",

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

1104 )

1105

1106 return event_to_pb(session, occurrence, context)

1107

1108 def ListMyEvents(

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

1110 ) -> events_pb2.ListMyEventsRes:

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

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

1113 page_token = (

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

1115 )

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

1117 page_number = request.page_number or 1

1118 # Calculate the offset for pagination

1119 offset = (page_number - 1) * page_size

1120 query = (

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

1122 )

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

1124

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

1126 include_subscribed = request.subscribed or include_all

1127 include_organizing = request.organizing or include_all

1128 include_attending = request.attending or include_all

1129 include_my_communities = request.my_communities or include_all

1130

1131 if include_attending and request.exclude_attending:

1132 context.abort_with_error_code(

1133 grpc.StatusCode.INVALID_ARGUMENT, "cannot_combine_attending_and_exclude_attending"

1134 )

1135

1136 where_ = []

1137

1138 if include_subscribed:

1139 query = query.outerjoin(

1140 EventSubscription,

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

1142 )

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

1144 if include_organizing:

1145 query = query.outerjoin(

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

1147 )

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

1149 if include_attending or request.exclude_attending:

1150 query = query.outerjoin(

1151 EventOccurrenceAttendee,

1152 and_(

1153 EventOccurrenceAttendee.occurrence_id == EventOccurrence.id,

1154 EventOccurrenceAttendee.user_id == context.user_id,

1155 ),

1156 )

1157 if include_attending:

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

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

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

1161 query = query.outerjoin(

1162 EventOrganizer,

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

1164 )

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

1166 if include_my_communities:

1167 my_communities = (

1168 session.execute(

1169 select(Node.id)

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

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

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

1173 .where(Cluster.is_official_cluster)

1174 .order_by(Node.id)

1175 .limit(100000)

1176 )

1177 .scalars()

1178 .all()

1179 )

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

1181

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

1183

1184 if request.my_communities_exclude_global:

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

1186

1187 if not request.include_cancelled:

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

1189

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

1191 cutoff = page_token - timedelta(seconds=1)

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

1193 else:

1194 cutoff = page_token + timedelta(seconds=1)

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

1196 # Count the total number of items for pagination

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

1198 # Apply pagination by page number

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

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

1201

1202 return events_pb2.ListMyEventsRes(

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

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

1205 total_items=total_items,

1206 )

1207

1208 def ListAllEvents(

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

1210 ) -> events_pb2.ListAllEventsRes:

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

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

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

1214

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

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

1217

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

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

1220

1221 if not request.past:

1222 cutoff = page_token - timedelta(seconds=1)

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

1224 else:

1225 cutoff = page_token + timedelta(seconds=1)

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

1227

1228 query = query.limit(page_size + 1)

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

1230

1231 return events_pb2.ListAllEventsRes(

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

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

1234 )

1235

1236 def InviteEventOrganizer(

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

1238 ) -> empty_pb2.Empty:

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

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

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

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

1243

1244 event, occurrence = res

1245

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

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

1248

1249 if occurrence.is_cancelled:

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

1251

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

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

1254

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

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

1257 ).scalar_one_or_none():

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

1259

1260 session.add(

1261 EventOrganizer(

1262 user_id=request.user_id,

1263 event_id=event.id,

1264 )

1265 )

1266 session.flush()

1267

1268 other_user_context = make_notification_user_context(user_id=request.user_id)

1269

1270 notify(

1271 session,

1272 user_id=request.user_id,

1273 topic_action=NotificationTopicAction.event__invite_organizer,

1274 key=str(event.id),

1275 data=notification_data_pb2.EventInviteOrganizer(

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

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

1278 ),

1279 )

1280

1281 return empty_pb2.Empty()

1282

1283 def RemoveEventOrganizer(

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

1285 ) -> empty_pb2.Empty:

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

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

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

1289

1290 event, occurrence = res

1291

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

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

1294

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

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

1297

1298 # Determine which user to remove

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

1300

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

1302 if event.owner_user_id == user_id_to_remove:

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

1304

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

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

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

1308

1309 # Find the organizer to remove

1310 organizer_to_remove = session.execute(

1311 select(EventOrganizer)

1312 .where(EventOrganizer.user_id == user_id_to_remove)

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

1314 ).scalar_one_or_none()

1315

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

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

1318

1319 session.delete(organizer_to_remove)

1320

1321 return empty_pb2.Empty()