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 # TODO(#9117): Remove update_str_items once known unused.

384 updated_str_items=payload.updated_str_items,

385 updated_enum_items=(

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

387 ),

388 ),

389 moderation_state_id=occurrence.moderation_state_id,

390 )

391

392

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

394 with session_scope() as session:

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

396

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

398

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

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

401

402 for user_id in set(subscribed_user_ids + attending_user_ids):

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

404 continue

405 context = make_notification_user_context(user_id=user_id)

406 notify(

407 session,

408 user_id=user_id,

409 topic_action=NotificationTopicAction.event__cancel,

410 key=str(payload.occurrence_id),

411 data=notification_data_pb2.EventCancel(

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

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

414 ),

415 moderation_state_id=occurrence.moderation_state_id,

416 )

417

418

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

420 with session_scope() as session:

421 event, occurrence = _get_event_and_occurrence_one(

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

423 )

424

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

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

427

428 for user_id in set(subscribed_user_ids + attending_user_ids):

429 context = make_notification_user_context(user_id=user_id)

430 notify(

431 session,

432 user_id=user_id,

433 topic_action=NotificationTopicAction.event__delete,

434 key=str(payload.occurrence_id),

435 data=notification_data_pb2.EventDelete(

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

437 ),

438 moderation_state_id=occurrence.moderation_state_id,

439 )

440

441

442class Events(events_pb2_grpc.EventsServicer):

443 def CreateEvent(

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

445 ) -> events_pb2.Event:

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

447 if not has_completed_profile(session, user):

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

449 if not request.title:

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

451 if not request.content:

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

453

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

455 timezone = _check_timezone_at(geom, context, session)

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

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

458 _check_occurrence_time_validity(start_datetime, end_datetime, context)

459

460 if request.parent_community_id:

461 parent_node = session.execute(

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

463 ).scalar_one_or_none()

464

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

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

467 else:

468 # parent community computed from geom

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

470

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

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

473

474 if (

475 request.photo_key

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

477 ):

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

479

480 thread = Thread()

481 session.add(thread)

482 session.flush()

483

484 event = Event(

485 title=request.title,

486 parent_node_id=parent_node.id,

487 owner_user_id=context.user_id,

488 thread_id=thread.id,

489 creator_user_id=context.user_id,

490 )

491 session.add(event)

492 session.flush()

493

494 occurrence: EventOccurrence | None = None

495

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

497 nonlocal occurrence

498 occurrence = EventOccurrence(

499 event_id=event.id,

500 content=request.content,

501 geom=geom,

502 address=address,

503 timezone=timezone.key,

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

505 during=TimestamptzRange(start_datetime, end_datetime),

506 creator_user_id=context.user_id,

507 moderation_state_id=moderation_state_id,

508 )

509 session.add(occurrence)

510 session.flush()

511 return occurrence.id

512

513 create_moderation(

514 session=session,

515 object_type=ModerationObjectType.event_occurrence,

516 object_id=create_occurrence,

517 creator_user_id=context.user_id,

518 )

519

520 assert occurrence is not None

521

522 session.add(

523 EventOrganizer(

524 user_id=context.user_id,

525 event_id=event.id,

526 )

527 )

528

529 session.add(

530 EventSubscription(

531 user_id=context.user_id,

532 event_id=event.id,

533 )

534 )

535

536 session.add(

537 EventOccurrenceAttendee(

538 user_id=context.user_id,

539 occurrence_id=occurrence.id,

540 attendee_status=AttendeeStatus.going,

541 )

542 )

543

544 session.commit()

545

546 log_event(

547 context,

548 session,

549 "event.created",

550 {

551 "event_id": event.id,

552 "occurrence_id": occurrence.id,

553 "parent_community_id": parent_node.id,

554 "parent_community_name": parent_node.official_cluster.name,

555 },

556 )

557

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

559 queue_job(

560 session,

561 job=generate_event_create_notifications,

562 payload=jobs_pb2.GenerateEventCreateNotificationsPayload(

563 inviting_user_id=user.id,

564 occurrence_id=occurrence.id,

565 approved=False,

566 ),

567 )

568

569 return event_to_pb(session, occurrence, context)

570

571 def ScheduleEvent(

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

573 ) -> events_pb2.Event:

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

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

576

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

578 timezone = _check_timezone_at(geom, context, session)

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

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

581 _check_occurrence_time_validity(start_datetime, end_datetime, context)

582

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

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

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

586

587 event, occurrence = res

588

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

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

591

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

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

594

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

596 request.photo_key

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

598 ):

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

600

601 during = TimestamptzRange(start_datetime, end_datetime)

602

603 # && is the overlap operator for ranges

604 if (

605 session.execute(

606 select(EventOccurrence.id)

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

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

609 .limit(1)

610 )

611 .scalars()

612 .one_or_none()

613 is not None

614 ):

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

616

617 new_occurrence: EventOccurrence | None = None

618

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

620 nonlocal new_occurrence

621 new_occurrence = EventOccurrence(

622 event_id=event.id,

623 content=request.content,

624 geom=geom,

625 address=address,

626 timezone=timezone.key,

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

628 during=during,

629 creator_user_id=context.user_id,

630 moderation_state_id=moderation_state_id,

631 )

632 session.add(new_occurrence)

633 session.flush()

634 return new_occurrence.id

635

636 create_moderation(

637 session=session,

638 object_type=ModerationObjectType.event_occurrence,

639 object_id=create_occurrence,

640 creator_user_id=context.user_id,

641 )

642

643 assert new_occurrence is not None

644

645 session.add(

646 EventOccurrenceAttendee(

647 user_id=context.user_id,

648 occurrence_id=new_occurrence.id,

649 attendee_status=AttendeeStatus.going,

650 )

651 )

652

653 session.flush()

654

655 # TODO: notify

656

657 return event_to_pb(session, new_occurrence, context)

658

659 def UpdateEvent(

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

661 ) -> events_pb2.Event:

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

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

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

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

666

667 event, occurrence = res

668

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

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

671

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

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

674

675 if occurrence.is_cancelled:

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

677

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

679

680 if request.HasField("title"):

681 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_TITLE)

682 event.title = request.title.value

683

684 if request.HasField("content"):

685 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_CONTENT)

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

687

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

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

690

691 old_timezone = ZoneInfo(occurrence.timezone)

692 timezone: ZoneInfo = old_timezone

693 if request.HasField("location"):

694 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_LOCATION)

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

696 timezone = _check_timezone_at(geom, context, session)

697 occurrence_update["geom"] = geom

698 occurrence_update["address"] = address

699 occurrence_update["timezone"] = timezone.key

700

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

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

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

704

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

706 start_datetime = _update_datetime(

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

708 timezone,

709 old_datetime=occurrence.start_time,

710 old_timezone=old_timezone,

711 context=context,

712 )

713 if start_datetime != occurrence.start_time:

714 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_START_TIME)

715

716 end_datetime = _update_datetime(

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

718 timezone,

719 old_datetime=occurrence.end_time,

720 old_timezone=old_timezone,

721 context=context,

722 )

723 if end_datetime != occurrence.end_time:

724 notify_updated.append(notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_END_TIME)

725

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

727 _check_occurrence_time_validity(start_datetime, end_datetime, context)

728

729 during = TimestamptzRange(start_datetime, end_datetime)

730

731 # && is the overlap operator for ranges

732 if (

733 session.execute(

734 select(EventOccurrence.id)

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

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

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

738 .limit(1)

739 )

740 .scalars()

741 .one_or_none()

742 is not None

743 ):

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

745

746 occurrence_update["during"] = during

747

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

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

750

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

752 if request.update_all_future:

753 session.execute(

754 update(EventOccurrence)

755 .where(EventOccurrence.end_time >= cutoff_time)

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

757 .values(occurrence_update)

758 .execution_options(synchronize_session=False)

759 )

760 else:

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

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

763 session.execute(

764 update(EventOccurrence)

765 .where(EventOccurrence.end_time >= cutoff_time)

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

767 .values(occurrence_update)

768 .execution_options(synchronize_session=False)

769 )

770

771 session.flush()

772

773 if notify_updated:

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

775 if request.should_notify:

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

777

778 queue_job(

779 session,

780 job=generate_event_update_notifications,

781 payload=jobs_pb2.GenerateEventUpdateNotificationsPayload(

782 updating_user_id=user.id,

783 occurrence_id=occurrence.id,

784 updated_enum_items=notify_updated,

785 ),

786 )

787 else:

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

789

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

791 session.refresh(occurrence)

792

793 return event_to_pb(session, occurrence, context)

794

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

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

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

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

799

800 if not occurrence:

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

802

803 return event_to_pb(session, occurrence, context)

804

805 def CancelEvent(

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

807 ) -> empty_pb2.Empty:

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

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

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

811

812 event, occurrence = res

813

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

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

816

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

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

819

820 occurrence.is_cancelled = True

821

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

823

824 queue_job(

825 session,

826 job=generate_event_cancel_notifications,

827 payload=jobs_pb2.GenerateEventCancelNotificationsPayload(

828 cancelling_user_id=context.user_id,

829 occurrence_id=occurrence.id,

830 ),

831 )

832

833 return empty_pb2.Empty()

834

835 def RequestCommunityInvite(

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

837 ) -> empty_pb2.Empty:

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

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

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

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

842

843 event, occurrence = res

844

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

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

847

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

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

850

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

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

853

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

855

856 if len(this_user_reqs) > 0:

857 context.abort_with_error_code(

858 grpc.StatusCode.FAILED_PRECONDITION, "event_community_invite_already_requested"

859 )

860

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

862

863 if len(approved_reqs) > 0:

864 context.abort_with_error_code(

865 grpc.StatusCode.FAILED_PRECONDITION, "event_community_invite_already_approved"

866 )

867

868 req = EventCommunityInviteRequest(

869 occurrence_id=request.event_id,

870 user_id=context.user_id,

871 )

872 session.add(req)

873 session.flush()

874

875 send_event_community_invite_request_email(session, req)

876

877 return empty_pb2.Empty()

878

879 def ListEventOccurrences(

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

881 ) -> events_pb2.ListEventOccurrencesRes:

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

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

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

885 initial_query = (

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

887 )

888 initial_query = where_moderated_content_visible(

889 initial_query, context, EventOccurrence, is_list_operation=False

890 )

891 occurrence = session.execute(initial_query).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

895 query = (

896 select(EventOccurrence)

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

898 .where(~EventOccurrence.is_deleted)

899 )

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

901

902 if not request.include_cancelled:

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

904

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

906 cutoff = page_token - timedelta(seconds=1)

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

908 else:

909 cutoff = page_token + timedelta(seconds=1)

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

911

912 query = query.limit(page_size + 1)

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

914

915 return events_pb2.ListEventOccurrencesRes(

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

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

918 )

919

920 def ListEventAttendees(

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

922 ) -> events_pb2.ListEventAttendeesRes:

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

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

925 occurrence = session.execute(

926 where_moderated_content_visible(

927 select(EventOccurrence)

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

929 .where(~EventOccurrence.is_deleted),

930 context,

931 EventOccurrence,

932 is_list_operation=False,

933 )

934 ).scalar_one_or_none()

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

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

937 attendees = (

938 session.execute(

939 where_users_column_visible(

940 select(EventOccurrenceAttendee)

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

942 .where(EventOccurrenceAttendee.user_id >= next_user_id)

943 .order_by(EventOccurrenceAttendee.user_id)

944 .limit(page_size + 1),

945 context,

946 EventOccurrenceAttendee.user_id,

947 )

948 )

949 .scalars()

950 .all()

951 )

952 return events_pb2.ListEventAttendeesRes(

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

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

955 )

956

957 def ListEventSubscribers(

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

959 ) -> events_pb2.ListEventSubscribersRes:

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

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

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

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

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

965 event, occurrence = res

966 subscribers = (

967 session.execute(

968 where_users_column_visible(

969 select(EventSubscription)

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

971 .where(EventSubscription.user_id >= next_user_id)

972 .order_by(EventSubscription.user_id)

973 .limit(page_size + 1),

974 context,

975 EventSubscription.user_id,

976 )

977 )

978 .scalars()

979 .all()

980 )

981 return events_pb2.ListEventSubscribersRes(

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

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

984 )

985

986 def ListEventOrganizers(

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

988 ) -> events_pb2.ListEventOrganizersRes:

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

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

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

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

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

994 event, occurrence = res

995 organizers = (

996 session.execute(

997 where_users_column_visible(

998 select(EventOrganizer)

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

1000 .where(EventOrganizer.user_id >= next_user_id)

1001 .order_by(EventOrganizer.user_id)

1002 .limit(page_size + 1),

1003 context,

1004 EventOrganizer.user_id,

1005 )

1006 )

1007 .scalars()

1008 .all()

1009 )

1010 return events_pb2.ListEventOrganizersRes(

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

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

1013 )

1014

1015 def TransferEvent(

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

1017 ) -> events_pb2.Event:

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

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

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

1021

1022 event, occurrence = res

1023

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

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

1026

1027 if occurrence.is_cancelled:

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

1029

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

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

1032

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

1034 cluster = session.execute(

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

1036 ).scalar_one_or_none()

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

1038 cluster = session.execute(

1039 select(Cluster)

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

1041 .where(Cluster.is_official_cluster)

1042 ).scalar_one_or_none()

1043

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

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

1046

1047 event.owner_user = None

1048 event.owner_cluster = cluster

1049

1050 session.commit()

1051 return event_to_pb(session, occurrence, context)

1052

1053 def SetEventSubscription(

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

1055 ) -> events_pb2.Event:

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

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

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

1059

1060 event, occurrence = res

1061

1062 if occurrence.is_cancelled:

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

1064

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

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

1067

1068 current_subscription = session.execute(

1069 select(EventSubscription)

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

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

1072 ).scalar_one_or_none()

1073

1074 # if not subscribed, subscribe

1075 if request.subscribe and not current_subscription:

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

1077

1078 # if subscribed but unsubbing, remove subscription

1079 if not request.subscribe and current_subscription:

1080 session.delete(current_subscription)

1081

1082 session.flush()

1083

1084 log_event(

1085 context,

1086 session,

1087 "event.subscription_set",

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

1089 )

1090

1091 return event_to_pb(session, occurrence, context)

1092

1093 def SetEventAttendance(

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

1095 ) -> events_pb2.Event:

1096 occurrence = session.execute(

1097 where_moderated_content_visible(

1098 select(EventOccurrence)

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

1100 .where(~EventOccurrence.is_deleted),

1101 context,

1102 EventOccurrence,

1103 is_list_operation=False,

1104 )

1105 ).scalar_one_or_none()

1106

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

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

1109

1110 if occurrence.is_cancelled:

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

1112

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

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

1115

1116 current_attendance = session.execute(

1117 select(EventOccurrenceAttendee)

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

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

1120 ).scalar_one_or_none()

1121

1122 if request.attendance_state == events_pb2.ATTENDANCE_STATE_NOT_GOING:

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

1124 session.delete(current_attendance)

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

1126 else:

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

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

1129 else:

1130 # create new

1131 attendance = EventOccurrenceAttendee(

1132 user_id=context.user_id,

1133 occurrence_id=occurrence.id,

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

1135 )

1136 session.add(attendance)

1137

1138 session.flush()

1139

1140 log_event(

1141 context,

1142 session,

1143 "event.attendance_set",

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

1145 )

1146

1147 return event_to_pb(session, occurrence, context)

1148

1149 def ListMyEvents(

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

1151 ) -> events_pb2.ListMyEventsRes:

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

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

1154 page_token = (

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

1156 )

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

1158 page_number = request.page_number or 1

1159 # Calculate the offset for pagination

1160 offset = (page_number - 1) * page_size

1161 query = (

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

1163 )

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

1165

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

1167 include_subscribed = request.subscribed or include_all

1168 include_organizing = request.organizing or include_all

1169 include_attending = request.attending or include_all

1170 include_my_communities = request.my_communities or include_all

1171

1172 if include_attending and request.exclude_attending:

1173 context.abort_with_error_code(

1174 grpc.StatusCode.INVALID_ARGUMENT, "cannot_combine_attending_and_exclude_attending"

1175 )

1176

1177 where_ = []

1178

1179 if include_subscribed:

1180 query = query.outerjoin(

1181 EventSubscription,

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

1183 )

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

1185 if include_organizing:

1186 query = query.outerjoin(

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

1188 )

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

1190 if include_attending or request.exclude_attending:

1191 query = query.outerjoin(

1192 EventOccurrenceAttendee,

1193 and_(

1194 EventOccurrenceAttendee.occurrence_id == EventOccurrence.id,

1195 EventOccurrenceAttendee.user_id == context.user_id,

1196 ),

1197 )

1198 if include_attending:

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

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

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

1202 query = query.outerjoin(

1203 EventOrganizer,

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

1205 )

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

1207 if include_my_communities:

1208 my_communities = (

1209 session.execute(

1210 select(Node.id)

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

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

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

1214 .where(Cluster.is_official_cluster)

1215 .order_by(Node.id)

1216 .limit(100000)

1217 )

1218 .scalars()

1219 .all()

1220 )

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

1222

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

1224

1225 if request.my_communities_exclude_global:

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

1227

1228 if not request.include_cancelled:

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

1230

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

1232 cutoff = page_token - timedelta(seconds=1)

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

1234 else:

1235 cutoff = page_token + timedelta(seconds=1)

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

1237 # Count the total number of items for pagination

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

1239 # Apply pagination by page number

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

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

1242

1243 return events_pb2.ListMyEventsRes(

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

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

1246 total_items=total_items,

1247 )

1248

1249 def ListAllEvents(

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

1251 ) -> events_pb2.ListAllEventsRes:

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

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

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

1255

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

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

1258

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

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

1261

1262 if not request.past:

1263 cutoff = page_token - timedelta(seconds=1)

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

1265 else:

1266 cutoff = page_token + timedelta(seconds=1)

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

1268

1269 query = query.limit(page_size + 1)

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

1271

1272 return events_pb2.ListAllEventsRes(

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

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

1275 )

1276

1277 def InviteEventOrganizer(

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

1279 ) -> empty_pb2.Empty:

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

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

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

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

1284

1285 event, occurrence = res

1286

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

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

1289

1290 if occurrence.is_cancelled:

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 if not session.execute( 1296 ↛ 1299line 1296 didn't jump to line 1299 because the condition on line 1296 was never true

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

1298 ).scalar_one_or_none():

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

1300

1301 session.add(

1302 EventOrganizer(

1303 user_id=request.user_id,

1304 event_id=event.id,

1305 )

1306 )

1307 session.flush()

1308

1309 other_user_context = make_notification_user_context(user_id=request.user_id)

1310

1311 notify(

1312 session,

1313 user_id=request.user_id,

1314 topic_action=NotificationTopicAction.event__invite_organizer,

1315 key=str(event.id),

1316 data=notification_data_pb2.EventInviteOrganizer(

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

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

1319 ),

1320 )

1321

1322 return empty_pb2.Empty()

1323

1324 def RemoveEventOrganizer(

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

1326 ) -> empty_pb2.Empty:

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

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

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

1330

1331 event, occurrence = res

1332

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

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

1335

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

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

1338

1339 # Determine which user to remove

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

1341

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

1343 if event.owner_user_id == user_id_to_remove:

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

1345

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

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

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

1349

1350 # Find the organizer to remove

1351 organizer_to_remove = session.execute(

1352 select(EventOrganizer)

1353 .where(EventOrganizer.user_id == user_id_to_remove)

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

1355 ).scalar_one_or_none()

1356

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

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

1359

1360 session.delete(organizer_to_remove)

1361

1362 return empty_pb2.Empty()