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

1import logging

2from datetime import datetime, timedelta

3from typing import Any, cast

4from zoneinfo import ZoneInfo

6import grpc

7from geoalchemy2 import WKBElement

8from google.protobuf import empty_pb2

9from psycopg.types.range import TimestamptzRange

10from sqlalchemy import Select, func, select

11from sqlalchemy.orm import Session

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

14from couchers.context import CouchersContext, make_notification_user_context

15from couchers.db import can_moderate_node, get_parent_node_at_location, session_scope

16from couchers.event_log import log_event

17from couchers.helpers.completed_profile import has_completed_profile

18from couchers.jobs.enqueue import queue_job

19from couchers.models import (

20 AttendeeStatus,

21 Cluster,

22 ClusterSubscription,

23 Event,

24 EventCommunityInviteRequest,

25 EventOccurrence,

26 EventOccurrenceAttendee,

27 EventOrganizer,

28 EventSubscription,

29 ModerationObjectType,

30 Node,

31 NodeType,

32 Thread,

33 Upload,

34 User,

35)

36from couchers.models.notifications import NotificationTopicAction

37from couchers.models.static import TimezoneArea

38from couchers.moderation.utils import create_moderation

39from couchers.notifications.notify import notify

40from couchers.proto import events_pb2, events_pb2_grpc, notification_data_pb2

41from couchers.proto.internal import jobs_pb2

42from couchers.servicers.api import user_model_to_pb

43from couchers.servicers.blocking import is_not_visible

44from couchers.servicers.threads import thread_to_pb

45from couchers.sql import users_visible, where_moderated_content_visible, where_users_column_visible

46from couchers.tasks import send_event_community_invite_request_email

47from couchers.utils import (

48 Timestamp_from_datetime,

49 create_coordinate,

50 datetime_to_iso8601_local,

51 dt_from_millis,

52 millis_from_dt,

53 not_none,

54 now,

55)

57logger = logging.getLogger(__name__)

59attendancestate2sql = {

60 events_pb2.AttendanceState.ATTENDANCE_STATE_NOT_GOING: None,

61 events_pb2.AttendanceState.ATTENDANCE_STATE_GOING: AttendeeStatus.going,

62}

64attendancestate2api = {

65 None: events_pb2.AttendanceState.ATTENDANCE_STATE_NOT_GOING,

66 AttendeeStatus.going: events_pb2.AttendanceState.ATTENDANCE_STATE_GOING,

67}

69MAX_PAGINATION_LENGTH = 25

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

73 """

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

75 """

76 if event.owner_user:

77 return event.owner_user_id == user_id

78 # otherwise owned by a cluster

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

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

83 """

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

85 """

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

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

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

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

92 return True

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

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

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

99 return (

100 _is_event_owner(event, user_id)

101 or _is_event_organizer(event, user_id)

102 or _can_moderate_event(session, event, user_id)

103 )

104

105

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

107 event = occurrence.event

108

109 next_occurrence = (

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

111 .order_by(EventOccurrence.end_time.asc())

112 .limit(1)

113 .one_or_none()

114 )

115

116 owner_community_id = None

117 owner_group_id = None

118 if event.owner_cluster:

119 if event.owner_cluster.is_official_cluster:

120 owner_community_id = event.owner_cluster.parent_node_id

121 else:

122 owner_group_id = event.owner_cluster.id

123

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

125 attendance_state = attendance.attendee_status if attendance else None

126

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

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

129

130 going_count = session.execute(

131 where_users_column_visible(

132 select(func.count())

133 .select_from(EventOccurrenceAttendee)

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

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

136 context,

137 EventOccurrenceAttendee.user_id,

138 )

139 ).scalar_one()

140 organizer_count = session.execute(

141 where_users_column_visible(

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

143 context,

144 EventOrganizer.user_id,

145 )

146 ).scalar_one()

147 subscriber_count = session.execute(

148 where_users_column_visible(

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

150 context,

151 EventSubscription.user_id,

152 )

153 ).scalar_one()

154

155 return events_pb2.Event(

156 event_id=occurrence.id,

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

158 is_cancelled=occurrence.is_cancelled,

159 is_deleted=occurrence.is_deleted,

160 title=event.title,

161 slug=event.slug,

162 content=occurrence.content,

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

164 photo_key=occurrence.photo_key or "",

165 location=events_pb2.EventLocation(

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

167 ),

168 created=Timestamp_from_datetime(occurrence.created),

169 last_edited=Timestamp_from_datetime(occurrence.last_edited),

170 creator_user_id=occurrence.creator_user_id,

171 start_time=Timestamp_from_datetime(occurrence.start_time),

172 end_time=Timestamp_from_datetime(occurrence.end_time),

173 timezone=occurrence.timezone,

174 attendance_state=attendancestate2api[attendance_state],

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

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

177 going_count=going_count,

178 organizer_count=organizer_count,

179 subscriber_count=subscriber_count,

180 owner_user_id=event.owner_user_id,

181 owner_community_id=owner_community_id,

182 owner_group_id=owner_group_id,

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

184 can_edit=can_edit,

185 can_moderate=can_moderate,

186 )

187

188

189def _get_event_and_occurrence_query(

190 occurrence_id: int,

191 include_deleted: bool,

192 context: CouchersContext | None = None,

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

194 query = (

195 select(Event, EventOccurrence)

196 .where(EventOccurrence.id == occurrence_id)

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

198 )

199

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

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

202

203 if context is not None:

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

205

206 return query

207

208

209def _get_event_and_occurrence_one(

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

211) -> tuple[Event, EventOccurrence]:

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

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

214 return result._tuple()

215

216

217def _get_event_and_occurrence_one_or_none(

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

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

220 result = session.execute(

221 _get_event_and_occurrence_query(occurrence_id, include_deleted, context=context)

222 ).one_or_none()

223 return result._tuple() if result else None

224

225

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

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

228 if not location or not location.address:

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

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

231 # No events allowed on Null Island

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

233

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

235 return (geom, location.address)

236

237

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

239 timezone_id = session.execute(

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

241 ).scalar_one_or_none()

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

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

244

245 return ZoneInfo(timezone_id)

246

247

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

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

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

251

252 try:

253 naive_datetime = datetime.fromisoformat(value)

254 except ValueError:

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

256

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

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

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

260

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

262

263

264def _update_datetime(

265 new_iso8601_local: str | None,

266 new_timezone: ZoneInfo,

267 old_datetime: datetime,

268 old_timezone: ZoneInfo,

269 context: CouchersContext,

270) -> datetime:

271 if new_iso8601_local is None and new_timezone != old_timezone:

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

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

274 if new_iso8601_local is None:

275 return old_datetime # No change

276 # New effective datetime/timestamp

277 return _check_iso8601_local_datetime(new_iso8601_local, new_timezone, context)

278

279

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

281 if start_time < now():

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

283 if end_time < start_time:

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

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

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

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

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

289

290

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

292 """

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

294 """

295 cluster = occurrence.event.parent_node.official_cluster

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

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

298 return [], occurrence.event.parent_node_id

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

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

301 else:

302 max_radius = 20000 # m

303 users = (

304 session.execute(

305 select(User)

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

307 .where(User.is_visible)

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

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

310 )

311 .scalars()

312 .all()

313 )

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

315

316

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

318 """

319 Background job to generated/fan out event notifications

320 """

321 # Import here to avoid circular dependency

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

323

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

325

326 with session_scope() as session:

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

328 creator = occurrence.creator_user

329

330 users, node_id = get_users_to_notify_for_new_event(session, occurrence)

331

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

333

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

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

336 return

337

338 for user in users:

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

340 continue

341 context = make_notification_user_context(user_id=user.id)

342 topic_action = (

343 NotificationTopicAction.event__create_approved

344 if payload.approved

345 else NotificationTopicAction.event__create_any

346 )

347 notify(

348 session,

349 user_id=user.id,

350 topic_action=topic_action,

351 key=str(payload.occurrence_id),

352 data=notification_data_pb2.EventCreate(

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

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

355 nearby=True if node_id is None else None,

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

357 ),

358 moderation_state_id=occurrence.moderation_state_id,

359 )

360

361

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

363 with session_scope() as session:

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

365

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

367

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

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

370

371 for user_id in set(subscribed_user_ids + attending_user_ids):

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

373 continue

374 context = make_notification_user_context(user_id=user_id)

375 notify(

376 session,

377 user_id=user_id,

378 topic_action=NotificationTopicAction.event__update,

379 key=str(payload.occurrence_id),

380 data=notification_data_pb2.EventUpdate(

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

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

383 updated_enum_items=(

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

385 ),

386 ),

387 moderation_state_id=occurrence.moderation_state_id,

388 )

389

390

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

392 with session_scope() as session:

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

394

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

396

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

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

399

400 for user_id in set(subscribed_user_ids + attending_user_ids):

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

402 continue

403 context = make_notification_user_context(user_id=user_id)

404 notify(

405 session,

406 user_id=user_id,

407 topic_action=NotificationTopicAction.event__cancel,

408 key=str(payload.occurrence_id),

409 data=notification_data_pb2.EventCancel(

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

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

412 ),

413 moderation_state_id=occurrence.moderation_state_id,

414 )

415

416

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

418 with session_scope() as session:

419 event, occurrence = _get_event_and_occurrence_one(

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

421 )

422

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

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

425

426 for user_id in set(subscribed_user_ids + attending_user_ids):

427 context = make_notification_user_context(user_id=user_id)

428 notify(

429 session,

430 user_id=user_id,

431 topic_action=NotificationTopicAction.event__delete,

432 key=str(payload.occurrence_id),

433 data=notification_data_pb2.EventDelete(

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

435 ),

436 moderation_state_id=occurrence.moderation_state_id,

437 )

438

439

440class Events(events_pb2_grpc.EventsServicer):

441 def CreateEvent(

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

443 ) -> events_pb2.Event:

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

445 if not has_completed_profile(session, user):

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

447 if not request.title:

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

449 if not request.content:

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

451

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

453 timezone = _check_timezone_at(geom, context, session)

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

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

456 _check_occurrence_time_validity(start_datetime, end_datetime, context)

457

458 if request.parent_community_id:

459 parent_node = session.execute(

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

461 ).scalar_one_or_none()

462

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

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

465 else:

466 # parent community computed from geom

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

468

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

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

471

472 if (

473 request.photo_key

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

475 ):

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

477

478 thread = Thread()

479 session.add(thread)

480 session.flush()

481

482 event = Event(

483 title=request.title,

484 parent_node_id=parent_node.id,

485 owner_user_id=context.user_id,

486 thread_id=thread.id,

487 creator_user_id=context.user_id,

488 )

489 session.add(event)

490 session.flush()

491

492 occurrence: EventOccurrence | None = None

493

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

495 nonlocal occurrence

496 occurrence = EventOccurrence(

497 event_id=event.id,

498 content=request.content,

499 geom=geom,

500 address=address,

501 timezone=timezone.key,

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

503 during=TimestamptzRange(start_datetime, end_datetime),

504 creator_user_id=context.user_id,

505 moderation_state_id=moderation_state_id,

506 )

507 session.add(occurrence)

508 session.flush()

509 return occurrence.id

510

511 create_moderation(

512 session=session,

513 object_type=ModerationObjectType.event_occurrence,

514 object_id=create_occurrence,

515 creator_user_id=context.user_id,

516 )

517

518 assert occurrence is not None

519

520 session.add(

521 EventOrganizer(

522 user_id=context.user_id,

523 event_id=event.id,

524 )

525 )

526

527 session.add(

528 EventSubscription(

529 user_id=context.user_id,

530 event_id=event.id,

531 )

532 )

533

534 session.add(

535 EventOccurrenceAttendee(

536 user_id=context.user_id,

537 occurrence_id=occurrence.id,

538 attendee_status=AttendeeStatus.going,

539 )

540 )

541

542 session.commit()

543

544 log_event(

545 context,

546 session,

547 "event.created",

548 {

549 "event_id": event.id,

550 "occurrence_id": occurrence.id,

551 "parent_community_id": parent_node.id,

552 "parent_community_name": parent_node.official_cluster.name,

553 },

554 )

555

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

557 queue_job(

558 session,

559 job=generate_event_create_notifications,

560 payload=jobs_pb2.GenerateEventCreateNotificationsPayload(

561 inviting_user_id=user.id,

562 occurrence_id=occurrence.id,

563 approved=False,

564 ),

565 )

566

567 return event_to_pb(session, occurrence, context)

568

569 def ScheduleEvent(

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

571 ) -> events_pb2.Event:

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

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

574

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

576 timezone = _check_timezone_at(geom, context, session)

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

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

579 _check_occurrence_time_validity(start_datetime, end_datetime, context)

580

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

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

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

584

585 event, occurrence = res

586

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

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

589

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

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

592

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

594 request.photo_key

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

596 ):

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

598

599 during = TimestamptzRange(start_datetime, end_datetime)

600

601 # && is the overlap operator for ranges

602 if (

603 session.execute(

604 select(EventOccurrence.id)

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

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

607 .limit(1)

608 )

609 .scalars()

610 .one_or_none()

611 is not None

612 ):

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

614

615 new_occurrence: EventOccurrence | None = None

616

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

618 nonlocal new_occurrence

619 new_occurrence = EventOccurrence(

620 event_id=event.id,

621 content=request.content,

622 geom=geom,

623 address=address,

624 timezone=timezone.key,

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

626 during=during,

627 creator_user_id=context.user_id,

628 moderation_state_id=moderation_state_id,

629 )

630 session.add(new_occurrence)

631 session.flush()

632 return new_occurrence.id

633

634 create_moderation(

635 session=session,

636 object_type=ModerationObjectType.event_occurrence,

637 object_id=create_occurrence,

638 creator_user_id=context.user_id,

639 )

640

641 assert new_occurrence is not None

642

643 session.add(

644 EventOccurrenceAttendee(

645 user_id=context.user_id,

646 occurrence_id=new_occurrence.id,

647 attendee_status=AttendeeStatus.going,

648 )

649 )

650

651 session.flush()

652

653 # TODO: notify

654

655 return event_to_pb(session, new_occurrence, context)

656

657 def UpdateEvent(

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

659 ) -> events_pb2.Event:

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

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

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

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

664

665 event, occurrence = res

666

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

669

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

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

672

673 if occurrence.is_cancelled:

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

675

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

677

678 if request.HasField("title"):

679 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_TITLE)

680 event.title = request.title.value

681

682 if request.HasField("content"):

683 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_CONTENT)

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

685

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

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

688

689 old_timezone = ZoneInfo(occurrence.timezone)

690 timezone: ZoneInfo = old_timezone

691 if request.HasField("location"):

692 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_LOCATION)

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

694 timezone = _check_timezone_at(geom, context, session)

695 occurrence_update["geom"] = geom

696 occurrence_update["address"] = address

697 occurrence_update["timezone"] = timezone.key

698

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

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

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

702

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

704 start_datetime = _update_datetime(

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

706 timezone,

707 old_datetime=occurrence.start_time,

708 old_timezone=old_timezone,

709 context=context,

710 )

711 if start_datetime != occurrence.start_time:

712 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_START_TIME)

713

714 end_datetime = _update_datetime(

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

716 timezone,

717 old_datetime=occurrence.end_time,

718 old_timezone=old_timezone,

719 context=context,

720 )

721 if end_datetime != occurrence.end_time:

722 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_END_TIME)

723

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

725 _check_occurrence_time_validity(start_datetime, end_datetime, context)

726

727 during = TimestamptzRange(start_datetime, end_datetime)

728

729 # && is the overlap operator for ranges

730 if (

731 session.execute(

732 select(EventOccurrence.id)

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

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

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

736 .limit(1)

737 )

738 .scalars()

739 .one_or_none()

740 is not None

741 ):

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

743

744 occurrence_update["during"] = during

745

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

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

748

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

750 if request.update_all_future:

751 session.execute(

752 update(EventOccurrence)

753 .where(EventOccurrence.end_time >= cutoff_time)

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

755 .values(occurrence_update)

756 .execution_options(synchronize_session=False)

757 )

758 else:

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

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

761 session.execute(

762 update(EventOccurrence)

763 .where(EventOccurrence.end_time >= cutoff_time)

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

765 .values(occurrence_update)

766 .execution_options(synchronize_session=False)

767 )

768

769 session.flush()

770

771 if notify_updated:

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

773 if request.should_notify:

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

775

776 queue_job(

777 session,

778 job=generate_event_update_notifications,

779 payload=jobs_pb2.GenerateEventUpdateNotificationsPayload(

780 updating_user_id=user.id,

781 occurrence_id=occurrence.id,

782 updated_enum_items=notify_updated,

783 ),

784 )

785 else:

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

787

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

789 session.refresh(occurrence)

790

791 return event_to_pb(session, occurrence, context)

792

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

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

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

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

797

798 if not occurrence:

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

800

801 return event_to_pb(session, occurrence, context)

802

803 def CancelEvent(

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

805 ) -> empty_pb2.Empty:

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

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

809

810 event, occurrence = res

811

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

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

814

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

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

817

818 occurrence.is_cancelled = True

819

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

821

822 queue_job(

823 session,

824 job=generate_event_cancel_notifications,

825 payload=jobs_pb2.GenerateEventCancelNotificationsPayload(

826 cancelling_user_id=context.user_id,

827 occurrence_id=occurrence.id,

828 ),

829 )

830

831 return empty_pb2.Empty()

832

833 def RequestCommunityInvite(

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

835 ) -> empty_pb2.Empty:

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

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

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

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

840

841 event, occurrence = res

842

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

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

845

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

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

848

849 if occurrence.end_time < now() - timedelta(hours=24): 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.FAILED_PRECONDITION, "event_cant_update_old_event")

851

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

853

854 if len(this_user_reqs) > 0:

855 context.abort_with_error_code(

856 grpc.StatusCode.FAILED_PRECONDITION, "event_community_invite_already_requested"

857 )

858

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

860

861 if len(approved_reqs) > 0:

862 context.abort_with_error_code(

863 grpc.StatusCode.FAILED_PRECONDITION, "event_community_invite_already_approved"

864 )

865

866 req = EventCommunityInviteRequest(

867 occurrence_id=request.event_id,

868 user_id=context.user_id,

869 )

870 session.add(req)

871 session.flush()

872

873 send_event_community_invite_request_email(session, req)

874

875 return empty_pb2.Empty()

876

877 def ListEventOccurrences(

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

879 ) -> events_pb2.ListEventOccurrencesRes:

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

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

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

883 initial_query = (

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

885 )

886 initial_query = where_moderated_content_visible(

887 initial_query, context, EventOccurrence, is_list_operation=False

888 )

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

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

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

892

893 query = (

894 select(EventOccurrence)

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

896 .where(~EventOccurrence.is_deleted)

897 )

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

899

900 if not request.include_cancelled:

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

902

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

904 cutoff = page_token - timedelta(seconds=1)

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

906 else:

907 cutoff = page_token + timedelta(seconds=1)

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

909

910 query = query.limit(page_size + 1)

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

912

913 return events_pb2.ListEventOccurrencesRes(

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

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

916 )

917

918 def ListEventAttendees(

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

920 ) -> events_pb2.ListEventAttendeesRes:

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

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

923 occurrence = session.execute(

924 where_moderated_content_visible(

925 select(EventOccurrence)

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

927 .where(~EventOccurrence.is_deleted),

928 context,

929 EventOccurrence,

930 is_list_operation=False,

931 )

932 ).scalar_one_or_none()

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

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

935 attendees = (

936 session.execute(

937 where_users_column_visible(

938 select(EventOccurrenceAttendee)

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

940 .where(EventOccurrenceAttendee.user_id >= next_user_id)

941 .order_by(EventOccurrenceAttendee.user_id)

942 .limit(page_size + 1),

943 context,

944 EventOccurrenceAttendee.user_id,

945 )

946 )

947 .scalars()

948 .all()

949 )

950 return events_pb2.ListEventAttendeesRes(

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

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

953 )

954

955 def ListEventSubscribers(

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

957 ) -> events_pb2.ListEventSubscribersRes:

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

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

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

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

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

963 event, occurrence = res

964 subscribers = (

965 session.execute(

966 where_users_column_visible(

967 select(EventSubscription)

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

969 .where(EventSubscription.user_id >= next_user_id)

970 .order_by(EventSubscription.user_id)

971 .limit(page_size + 1),

972 context,

973 EventSubscription.user_id,

974 )

975 )

976 .scalars()

977 .all()

978 )

979 return events_pb2.ListEventSubscribersRes(

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

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

982 )

983

984 def ListEventOrganizers(

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

986 ) -> events_pb2.ListEventOrganizersRes:

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

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

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

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

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

992 event, occurrence = res

993 organizers = (

994 session.execute(

995 where_users_column_visible(

996 select(EventOrganizer)

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

998 .where(EventOrganizer.user_id >= next_user_id)

999 .order_by(EventOrganizer.user_id)

1000 .limit(page_size + 1),

1001 context,

1002 EventOrganizer.user_id,

1003 )

1004 )

1005 .scalars()

1006 .all()

1007 )

1008 return events_pb2.ListEventOrganizersRes(

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

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

1011 )

1012

1013 def TransferEvent(

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

1015 ) -> events_pb2.Event:

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

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

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

1019

1020 event, occurrence = res

1021

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

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

1024

1025 if occurrence.is_cancelled:

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

1027

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

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

1030

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

1032 cluster = session.execute(

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

1034 ).scalar_one_or_none()

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

1036 cluster = session.execute(

1037 select(Cluster)

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

1039 .where(Cluster.is_official_cluster)

1040 ).scalar_one_or_none()

1041

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

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

1044

1045 event.owner_user = None

1046 event.owner_cluster = cluster

1047

1048 session.commit()

1049 return event_to_pb(session, occurrence, context)

1050

1051 def SetEventSubscription(

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

1053 ) -> events_pb2.Event:

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

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

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

1057

1058 event, occurrence = res

1059

1060 if occurrence.is_cancelled:

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

1062

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

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

1065

1066 current_subscription = session.execute(

1067 select(EventSubscription)

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

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

1070 ).scalar_one_or_none()

1071

1072 # if not subscribed, subscribe

1073 if request.subscribe and not current_subscription:

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

1075

1076 # if subscribed but unsubbing, remove subscription

1077 if not request.subscribe and current_subscription:

1078 session.delete(current_subscription)

1079

1080 session.flush()

1081

1082 log_event(

1083 context,

1084 session,

1085 "event.subscription_set",

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

1087 )

1088

1089 return event_to_pb(session, occurrence, context)

1090

1091 def SetEventAttendance(

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

1093 ) -> events_pb2.Event:

1094 occurrence = session.execute(

1095 where_moderated_content_visible(

1096 select(EventOccurrence)

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

1098 .where(~EventOccurrence.is_deleted),

1099 context,

1100 EventOccurrence,

1101 is_list_operation=False,

1102 )

1103 ).scalar_one_or_none()

1104

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

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

1107

1108 if occurrence.is_cancelled:

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

1110

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

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

1113

1114 current_attendance = session.execute(

1115 select(EventOccurrenceAttendee)

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

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

1118 ).scalar_one_or_none()

1119

1120 if request.attendance_state == events_pb2.ATTENDANCE_STATE_NOT_GOING:

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

1122 session.delete(current_attendance)

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

1124 else:

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

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

1127 else:

1128 # create new

1129 attendance = EventOccurrenceAttendee(

1130 user_id=context.user_id,

1131 occurrence_id=occurrence.id,

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

1133 )

1134 session.add(attendance)

1135

1136 session.flush()

1137

1138 log_event(

1139 context,

1140 session,

1141 "event.attendance_set",

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

1143 )

1144

1145 return event_to_pb(session, occurrence, context)

1146

1147 def ListMyEvents(

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

1149 ) -> events_pb2.ListMyEventsRes:

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

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

1152 page_token = (

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

1154 )

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

1156 page_number = request.page_number or 1

1157 # Calculate the offset for pagination

1158 offset = (page_number - 1) * page_size

1159 query = (

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

1161 )

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

1163

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

1165 include_subscribed = request.subscribed or include_all

1166 include_organizing = request.organizing or include_all

1167 include_attending = request.attending or include_all

1168 include_my_communities = request.my_communities or include_all

1169

1170 if include_attending and request.exclude_attending:

1171 context.abort_with_error_code(

1172 grpc.StatusCode.INVALID_ARGUMENT, "cannot_combine_attending_and_exclude_attending"

1173 )

1174

1175 where_ = []

1176

1177 if include_subscribed:

1178 query = query.outerjoin(

1179 EventSubscription,

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

1181 )

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

1183 if include_organizing:

1184 query = query.outerjoin(

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

1186 )

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

1188 if include_attending or request.exclude_attending:

1189 query = query.outerjoin(

1190 EventOccurrenceAttendee,

1191 and_(

1192 EventOccurrenceAttendee.occurrence_id == EventOccurrence.id,

1193 EventOccurrenceAttendee.user_id == context.user_id,

1194 ),

1195 )

1196 if include_attending:

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

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

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

1200 query = query.outerjoin(

1201 EventOrganizer,

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

1203 )

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

1205 if include_my_communities:

1206 my_communities = (

1207 session.execute(

1208 select(Node.id)

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

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

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

1212 .where(Cluster.is_official_cluster)

1213 .order_by(Node.id)

1214 .limit(100000)

1215 )

1216 .scalars()

1217 .all()

1218 )

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

1220

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

1222

1223 if request.my_communities_exclude_global:

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

1225

1226 if not request.include_cancelled:

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

1228

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

1230 cutoff = page_token - timedelta(seconds=1)

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

1232 else:

1233 cutoff = page_token + timedelta(seconds=1)

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

1235 # Count the total number of items for pagination

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

1237 # Apply pagination by page number

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

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

1240

1241 return events_pb2.ListMyEventsRes(

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

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

1244 total_items=total_items,

1245 )

1246

1247 def ListAllEvents(

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

1249 ) -> events_pb2.ListAllEventsRes:

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

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

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

1253

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

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

1256

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

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

1259

1260 if not request.past:

1261 cutoff = page_token - timedelta(seconds=1)

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

1263 else:

1264 cutoff = page_token + timedelta(seconds=1)

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

1266

1267 query = query.limit(page_size + 1)

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

1269

1270 return events_pb2.ListAllEventsRes(

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

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

1273 )

1274

1275 def InviteEventOrganizer(

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

1277 ) -> empty_pb2.Empty:

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

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

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

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

1282

1283 event, occurrence = res

1284

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

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

1287

1288 if occurrence.is_cancelled:

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

1290

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

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

1293

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

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

1296 ).scalar_one_or_none():

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

1298

1299 session.add(

1300 EventOrganizer(

1301 user_id=request.user_id,

1302 event_id=event.id,

1303 )

1304 )

1305 session.flush()

1306

1307 other_user_context = make_notification_user_context(user_id=request.user_id)

1308

1309 notify(

1310 session,

1311 user_id=request.user_id,

1312 topic_action=NotificationTopicAction.event__invite_organizer,

1313 key=str(event.id),

1314 data=notification_data_pb2.EventInviteOrganizer(

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

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

1317 ),

1318 )

1319

1320 return empty_pb2.Empty()

1321

1322 def RemoveEventOrganizer(

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

1324 ) -> empty_pb2.Empty:

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

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

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

1328

1329 event, occurrence = res

1330

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

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

1333

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

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

1336

1337 # Determine which user to remove

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

1339

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

1341 if event.owner_user_id == user_id_to_remove:

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

1343

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

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

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

1347

1348 # Find the organizer to remove

1349 organizer_to_remove = session.execute(

1350 select(EventOrganizer)

1351 .where(EventOrganizer.user_id == user_id_to_remove)

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

1353 ).scalar_one_or_none()

1354

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

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

1357

1358 session.delete(organizer_to_remove)

1359

1360 return empty_pb2.Empty()