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 return events_pb2.Event(

153 event_id=occurrence.id,

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

155 is_cancelled=occurrence.is_cancelled,

156 is_deleted=occurrence.is_deleted,

157 title=event.title,

158 slug=event.slug,

159 content=occurrence.content,

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

161 photo_key=occurrence.photo_key or "",

162 location=events_pb2.EventLocation(

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

164 ),

165 created=Timestamp_from_datetime(occurrence.created),

166 last_edited=Timestamp_from_datetime(occurrence.last_edited),

167 creator_user_id=occurrence.creator_user_id,

168 start_time=Timestamp_from_datetime(occurrence.start_time),

169 end_time=Timestamp_from_datetime(occurrence.end_time),

170 timezone=occurrence.timezone,

171 attendance_state=attendancestate2api[attendance_state],

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

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

174 going_count=going_count,

175 organizer_count=organizer_count,

176 subscriber_count=subscriber_count,

177 owner_user_id=event.owner_user_id,

178 owner_community_id=owner_community_id,

179 owner_group_id=owner_group_id,

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

181 can_edit=can_edit,

182 can_moderate=can_moderate,

183 )

184

185

186def _get_event_and_occurrence_query(

187 occurrence_id: int,

188 include_deleted: bool,

189 context: CouchersContext | None = None,

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

191 query = (

192 select(Event, EventOccurrence)

193 .where(EventOccurrence.id == occurrence_id)

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

195 )

196

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

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

199

200 if context is not None:

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

202

203 return query

204

205

206def _get_event_and_occurrence_one(

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

208) -> tuple[Event, EventOccurrence]:

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

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

211 return result._tuple()

212

213

214def _get_event_and_occurrence_one_or_none(

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

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

217 result = session.execute(

218 _get_event_and_occurrence_query(occurrence_id, include_deleted, context=context)

219 ).one_or_none()

220 return result._tuple() if result else None

221

222

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

224 if start_time < now():

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

226 if end_time < start_time:

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

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

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

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

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

232

233

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

235 """

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

237 """

238 cluster = occurrence.event.parent_node.official_cluster

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

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

241 return [], occurrence.event.parent_node_id

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

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

244 else:

245 max_radius = 20000 # m

246 users = (

247 session.execute(

248 select(User)

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

250 .where(User.is_visible)

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

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

253 )

254 .scalars()

255 .all()

256 )

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

258

259

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

261 """

262 Background job to generated/fan out event notifications

263 """

264 # Import here to avoid circular dependency

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

266

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

268

269 with session_scope() as session:

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

271 creator = occurrence.creator_user

272

273 users, node_id = get_users_to_notify_for_new_event(session, occurrence)

274

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

276

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

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

279 return

280

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

282 already_involved_user_ids = set(

283 session.execute(

284 select(EventOccurrenceAttendee.user_id).where(EventOccurrenceAttendee.occurrence_id == occurrence.id)

285 ).scalars()

286 ) | set(session.execute(select(EventOrganizer.user_id).where(EventOrganizer.event_id == event.id)).scalars())

287

288 for user in users:

289 if user.id in already_involved_user_ids:

290 continue

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

292 continue

293 context = make_notification_user_context(user_id=user.id)

294 topic_action = (

295 NotificationTopicAction.event__create_approved

296 if payload.approved

297 else NotificationTopicAction.event__create_any

298 )

299 notify(

300 session,

301 user_id=user.id,

302 topic_action=topic_action,

303 key=str(payload.occurrence_id),

304 data=notification_data_pb2.EventCreate(

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

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

307 nearby=True if node_id is None else None,

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

309 ),

310 moderation_state_id=occurrence.moderation_state_id,

311 )

312

313

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

315 with session_scope() as session:

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

317

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

319

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

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

322

323 for user_id in set(subscribed_user_ids + attending_user_ids):

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

325 continue

326 context = make_notification_user_context(user_id=user_id)

327 notify(

328 session,

329 user_id=user_id,

330 topic_action=NotificationTopicAction.event__update,

331 key=str(payload.occurrence_id),

332 data=notification_data_pb2.EventUpdate(

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

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

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

336 updated_str_items=payload.updated_str_items,

337 updated_enum_items=(

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

339 ),

340 ),

341 moderation_state_id=occurrence.moderation_state_id,

342 )

343

344

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

346 with session_scope() as session:

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

348

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

350

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

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

353

354 for user_id in set(subscribed_user_ids + attending_user_ids):

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

356 continue

357 context = make_notification_user_context(user_id=user_id)

358 notify(

359 session,

360 user_id=user_id,

361 topic_action=NotificationTopicAction.event__cancel,

362 key=str(payload.occurrence_id),

363 data=notification_data_pb2.EventCancel(

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

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

366 ),

367 moderation_state_id=occurrence.moderation_state_id,

368 )

369

370

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

372 with session_scope() as session:

373 event, occurrence = _get_event_and_occurrence_one(

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

375 )

376

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

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

379

380 for user_id in set(subscribed_user_ids + attending_user_ids):

381 context = make_notification_user_context(user_id=user_id)

382 notify(

383 session,

384 user_id=user_id,

385 topic_action=NotificationTopicAction.event__delete,

386 key=str(payload.occurrence_id),

387 data=notification_data_pb2.EventDelete(

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

389 ),

390 moderation_state_id=occurrence.moderation_state_id,

391 )

392

393

394class Events(events_pb2_grpc.EventsServicer):

395 def CreateEvent(

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

397 ) -> events_pb2.Event:

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

399 if not has_completed_profile(session, user):

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

401 if not request.title:

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

403 if not request.content:

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

405

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

407 if not (

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

409 ):

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

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

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

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

414 address = request.location.address

415

416 start_time = to_aware_datetime(request.start_time)

417 end_time = to_aware_datetime(request.end_time)

418

419 _check_occurrence_time_validity(start_time, end_time, context)

420

421 if request.parent_community_id:

422 parent_node = session.execute(

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

424 ).scalar_one_or_none()

425

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

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

428 else:

429 # parent community computed from geom

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

431

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

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

434

435 if (

436 request.photo_key

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

438 ):

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

440

441 thread = Thread()

442 session.add(thread)

443 session.flush()

444

445 event = Event(

446 title=request.title,

447 parent_node_id=parent_node.id,

448 owner_user_id=context.user_id,

449 thread_id=thread.id,

450 creator_user_id=context.user_id,

451 )

452 session.add(event)

453 session.flush()

454

455 occurrence: EventOccurrence | None = None

456

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

458 nonlocal occurrence

459 occurrence = EventOccurrence(

460 event_id=event.id,

461 content=request.content,

462 geom=geom,

463 address=address,

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 photo_key=request.photo_key if request.photo_key != "" else None,

595 # timezone=timezone,

596 during=during,

597 creator_user_id=context.user_id,

598 moderation_state_id=moderation_state_id,

599 )

600 session.add(new_occurrence)

601 session.flush()

602 return new_occurrence.id

603

604 create_moderation(

605 session=session,

606 object_type=ModerationObjectType.event_occurrence,

607 object_id=create_occurrence,

608 creator_user_id=context.user_id,

609 )

610

611 assert new_occurrence is not None

612

613 session.add(

614 EventOccurrenceAttendee(

615 user_id=context.user_id,

616 occurrence_id=new_occurrence.id,

617 attendee_status=AttendeeStatus.going,

618 )

619 )

620

621 session.flush()

622

623 # TODO: notify

624

625 return event_to_pb(session, new_occurrence, context)

626

627 def UpdateEvent(

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

629 ) -> events_pb2.Event:

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

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

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

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

634

635 event, occurrence = res

636

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

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

639

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

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

642

643 if occurrence.is_cancelled:

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

645

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

647

648 if request.HasField("title"):

649 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_TITLE)

650 event.title = request.title.value

651

652 if request.HasField("content"):

653 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_CONTENT)

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

655

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

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

658

659 if request.HasField("location"):

660 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_LOCATION)

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

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

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

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

665

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

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

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

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

670 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_START_TIME)

671 start_time = to_aware_datetime(request.start_time)

672 else:

673 start_time = occurrence.start_time

674 if request.HasField("end_time"):

675 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_END_TIME)

676 end_time = to_aware_datetime(request.end_time)

677 else:

678 end_time = occurrence.end_time

679

680 _check_occurrence_time_validity(start_time, end_time, context)

681

682 during = TimestamptzRange(start_time, end_time)

683

684 # && is the overlap operator for ranges

685 if (

686 session.execute(

687 select(EventOccurrence.id)

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

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

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

691 .limit(1)

692 )

693 .scalars()

694 .one_or_none()

695 is not None

696 ):

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

698

699 occurrence_update["during"] = during

700

701 # TODO

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

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

704

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

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

707

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

709 if request.update_all_future:

710 session.execute(

711 update(EventOccurrence)

712 .where(EventOccurrence.end_time >= cutoff_time)

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

714 .values(occurrence_update)

715 .execution_options(synchronize_session=False)

716 )

717 else:

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

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

720 session.execute(

721 update(EventOccurrence)

722 .where(EventOccurrence.end_time >= cutoff_time)

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

724 .values(occurrence_update)

725 .execution_options(synchronize_session=False)

726 )

727

728 session.flush()

729

730 if notify_updated:

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

732 if request.should_notify:

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

734

735 queue_job(

736 session,

737 job=generate_event_update_notifications,

738 payload=jobs_pb2.GenerateEventUpdateNotificationsPayload(

739 updating_user_id=user.id,

740 occurrence_id=occurrence.id,

741 updated_enum_items=notify_updated,

742 ),

743 )

744 else:

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

746

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

748 session.refresh(occurrence)

749

750 return event_to_pb(session, occurrence, context)

751

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

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

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

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

756

757 if not occurrence:

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

759

760 return event_to_pb(session, occurrence, context)

761

762 def CancelEvent(

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

764 ) -> empty_pb2.Empty:

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

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

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

768

769 event, occurrence = res

770

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

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

773

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

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

776

777 occurrence.is_cancelled = True

778

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

780

781 queue_job(

782 session,

783 job=generate_event_cancel_notifications,

784 payload=jobs_pb2.GenerateEventCancelNotificationsPayload(

785 cancelling_user_id=context.user_id,

786 occurrence_id=occurrence.id,

787 ),

788 )

789

790 return empty_pb2.Empty()

791

792 def RequestCommunityInvite(

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

794 ) -> empty_pb2.Empty:

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

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

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

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

799

800 event, occurrence = res

801

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

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

804

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

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

807

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

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

810

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

812

813 if len(this_user_reqs) > 0:

814 context.abort_with_error_code(

815 grpc.StatusCode.FAILED_PRECONDITION, "event_community_invite_already_requested"

816 )

817

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

819

820 if len(approved_reqs) > 0:

821 context.abort_with_error_code(

822 grpc.StatusCode.FAILED_PRECONDITION, "event_community_invite_already_approved"

823 )

824

825 req = EventCommunityInviteRequest(

826 occurrence_id=request.event_id,

827 user_id=context.user_id,

828 )

829 session.add(req)

830 session.flush()

831

832 send_event_community_invite_request_email(session, req)

833

834 return empty_pb2.Empty()

835

836 def ListEventOccurrences(

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

838 ) -> events_pb2.ListEventOccurrencesRes:

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

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

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

842 initial_query = (

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

844 )

845 initial_query = where_moderated_content_visible(

846 initial_query, context, EventOccurrence, is_list_operation=False

847 )

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

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

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

851

852 query = (

853 select(EventOccurrence)

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

855 .where(~EventOccurrence.is_deleted)

856 )

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

858

859 if not request.include_cancelled:

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

861

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

863 cutoff = page_token - timedelta(seconds=1)

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

865 else:

866 cutoff = page_token + timedelta(seconds=1)

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

868

869 query = query.limit(page_size + 1)

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

871

872 return events_pb2.ListEventOccurrencesRes(

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

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

875 )

876

877 def ListEventAttendees(

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

879 ) -> events_pb2.ListEventAttendeesRes:

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

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

882 occurrence = session.execute(

883 where_moderated_content_visible(

884 select(EventOccurrence)

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

886 .where(~EventOccurrence.is_deleted),

887 context,

888 EventOccurrence,

889 is_list_operation=False,

890 )

891 ).scalar_one_or_none()

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

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

894 attendees = (

895 session.execute(

896 where_users_column_visible(

897 select(EventOccurrenceAttendee)

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

899 .where(EventOccurrenceAttendee.user_id >= next_user_id)

900 .order_by(EventOccurrenceAttendee.user_id)

901 .limit(page_size + 1),

902 context,

903 EventOccurrenceAttendee.user_id,

904 )

905 )

906 .scalars()

907 .all()

908 )

909 return events_pb2.ListEventAttendeesRes(

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

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

912 )

913

914 def ListEventSubscribers(

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

916 ) -> events_pb2.ListEventSubscribersRes:

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

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

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

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

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

922 event, occurrence = res

923 subscribers = (

924 session.execute(

925 where_users_column_visible(

926 select(EventSubscription)

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

928 .where(EventSubscription.user_id >= next_user_id)

929 .order_by(EventSubscription.user_id)

930 .limit(page_size + 1),

931 context,

932 EventSubscription.user_id,

933 )

934 )

935 .scalars()

936 .all()

937 )

938 return events_pb2.ListEventSubscribersRes(

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

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

941 )

942

943 def ListEventOrganizers(

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

945 ) -> events_pb2.ListEventOrganizersRes:

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

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

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

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

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

951 event, occurrence = res

952 organizers = (

953 session.execute(

954 where_users_column_visible(

955 select(EventOrganizer)

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

957 .where(EventOrganizer.user_id >= next_user_id)

958 .order_by(EventOrganizer.user_id)

959 .limit(page_size + 1),

960 context,

961 EventOrganizer.user_id,

962 )

963 )

964 .scalars()

965 .all()

966 )

967 return events_pb2.ListEventOrganizersRes(

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

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

970 )

971

972 def TransferEvent(

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

974 ) -> events_pb2.Event:

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

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

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

978

979 event, occurrence = res

980

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

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

983

984 if occurrence.is_cancelled:

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

986

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

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

989

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

991 cluster = session.execute(

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

993 ).scalar_one_or_none()

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

995 cluster = session.execute(

996 select(Cluster)

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

998 .where(Cluster.is_official_cluster)

999 ).scalar_one_or_none()

1000

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

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

1003

1004 event.owner_user = None

1005 event.owner_cluster = cluster

1006

1007 session.commit()

1008 return event_to_pb(session, occurrence, context)

1009

1010 def SetEventSubscription(

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

1012 ) -> events_pb2.Event:

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

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

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

1016

1017 event, occurrence = res

1018

1019 if occurrence.is_cancelled:

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

1021

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

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

1024

1025 current_subscription = session.execute(

1026 select(EventSubscription)

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

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

1029 ).scalar_one_or_none()

1030

1031 # if not subscribed, subscribe

1032 if request.subscribe and not current_subscription:

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

1034

1035 # if subscribed but unsubbing, remove subscription

1036 if not request.subscribe and current_subscription:

1037 session.delete(current_subscription)

1038

1039 session.flush()

1040

1041 log_event(

1042 context,

1043 session,

1044 "event.subscription_set",

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

1046 )

1047

1048 return event_to_pb(session, occurrence, context)

1049

1050 def SetEventAttendance(

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

1052 ) -> events_pb2.Event:

1053 occurrence = session.execute(

1054 where_moderated_content_visible(

1055 select(EventOccurrence)

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

1057 .where(~EventOccurrence.is_deleted),

1058 context,

1059 EventOccurrence,

1060 is_list_operation=False,

1061 )

1062 ).scalar_one_or_none()

1063

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

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

1066

1067 if occurrence.is_cancelled:

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

1069

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

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

1072

1073 current_attendance = session.execute(

1074 select(EventOccurrenceAttendee)

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

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

1077 ).scalar_one_or_none()

1078

1079 if request.attendance_state == events_pb2.ATTENDANCE_STATE_NOT_GOING:

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

1081 session.delete(current_attendance)

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

1083 else:

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

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

1086 else:

1087 # create new

1088 attendance = EventOccurrenceAttendee(

1089 user_id=context.user_id,

1090 occurrence_id=occurrence.id,

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

1092 )

1093 session.add(attendance)

1094

1095 session.flush()

1096

1097 log_event(

1098 context,

1099 session,

1100 "event.attendance_set",

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

1102 )

1103

1104 return event_to_pb(session, occurrence, context)

1105

1106 def ListMyEvents(

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

1108 ) -> events_pb2.ListMyEventsRes:

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

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

1111 page_token = (

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

1113 )

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

1115 page_number = request.page_number or 1

1116 # Calculate the offset for pagination

1117 offset = (page_number - 1) * page_size

1118 query = (

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

1120 )

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

1122

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

1124 include_subscribed = request.subscribed or include_all

1125 include_organizing = request.organizing or include_all

1126 include_attending = request.attending or include_all

1127 include_my_communities = request.my_communities or include_all

1128

1129 if include_attending and request.exclude_attending:

1130 context.abort_with_error_code(

1131 grpc.StatusCode.INVALID_ARGUMENT, "cannot_combine_attending_and_exclude_attending"

1132 )

1133

1134 where_ = []

1135

1136 if include_subscribed:

1137 query = query.outerjoin(

1138 EventSubscription,

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

1140 )

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

1142 if include_organizing:

1143 query = query.outerjoin(

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

1145 )

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

1147 if include_attending or request.exclude_attending:

1148 query = query.outerjoin(

1149 EventOccurrenceAttendee,

1150 and_(

1151 EventOccurrenceAttendee.occurrence_id == EventOccurrence.id,

1152 EventOccurrenceAttendee.user_id == context.user_id,

1153 ),

1154 )

1155 if include_attending:

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

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

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

1159 query = query.outerjoin(

1160 EventOrganizer,

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

1162 )

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

1164 if include_my_communities:

1165 my_communities = (

1166 session.execute(

1167 select(Node.id)

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

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

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

1171 .where(Cluster.is_official_cluster)

1172 .order_by(Node.id)

1173 .limit(100000)

1174 )

1175 .scalars()

1176 .all()

1177 )

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

1179

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

1181

1182 if request.my_communities_exclude_global:

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

1184

1185 if not request.include_cancelled:

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

1187

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

1189 cutoff = page_token - timedelta(seconds=1)

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

1191 else:

1192 cutoff = page_token + timedelta(seconds=1)

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

1194 # Count the total number of items for pagination

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

1196 # Apply pagination by page number

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

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

1199

1200 return events_pb2.ListMyEventsRes(

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

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

1203 total_items=total_items,

1204 )

1205

1206 def ListAllEvents(

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

1208 ) -> events_pb2.ListAllEventsRes:

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

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

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

1212

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

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

1215

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

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

1218

1219 if not request.past:

1220 cutoff = page_token - timedelta(seconds=1)

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

1222 else:

1223 cutoff = page_token + timedelta(seconds=1)

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

1225

1226 query = query.limit(page_size + 1)

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

1228

1229 return events_pb2.ListAllEventsRes(

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

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

1232 )

1233

1234 def InviteEventOrganizer(

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

1236 ) -> empty_pb2.Empty:

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

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

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

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

1241

1242 event, occurrence = res

1243

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

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

1246

1247 if occurrence.is_cancelled:

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

1249

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

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

1252

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

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

1255 ).scalar_one_or_none():

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

1257

1258 session.add(

1259 EventOrganizer(

1260 user_id=request.user_id,

1261 event_id=event.id,

1262 )

1263 )

1264 session.flush()

1265

1266 other_user_context = make_notification_user_context(user_id=request.user_id)

1267

1268 notify(

1269 session,

1270 user_id=request.user_id,

1271 topic_action=NotificationTopicAction.event__invite_organizer,

1272 key=str(event.id),

1273 data=notification_data_pb2.EventInviteOrganizer(

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

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

1276 ),

1277 )

1278

1279 return empty_pb2.Empty()

1280

1281 def RemoveEventOrganizer(

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

1283 ) -> empty_pb2.Empty:

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

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

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

1287

1288 event, occurrence = res

1289

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

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

1292

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

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

1295

1296 # Determine which user to remove

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

1298

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

1300 if event.owner_user_id == user_id_to_remove:

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

1302

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

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

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

1306

1307 # Find the organizer to remove

1308 organizer_to_remove = session.execute(

1309 select(EventOrganizer)

1310 .where(EventOrganizer.user_id == user_id_to_remove)

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

1312 ).scalar_one_or_none()

1313

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

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

1316

1317 session.delete(organizer_to_remove)

1318

1319 return empty_pb2.Empty()