Coverage for functions\flipdare\service\notification

1#!/usr/bin/env python

4# This file is part of Flipdare's proprietary software and contains

5# confidential and copyrighted material. Unauthorised copying,

6# modification, distribution, or use of this file is strictly

7# prohibited without prior written permission from Flipdare Pty Ltd.

9# This software includes third-party components licensed under MIT,

10# BSD, and Apache 2.0 licences. See THIRD_PARTY_NOTICES for details.

11#

13from __future__ import annotations

14from typing import TYPE_CHECKING

15from flipdare.app_log import LOG

16from flipdare.constants import IS_DEBUG, NO_DOC_ID

17from flipdare.message.user_message import NotificationMessage

18from flipdare.result.app_result import AppResult

19from flipdare.result.job_result import JobResult

20from flipdare.core.trigger_decorator import trigger_decorator

21from flipdare.service._service_provider import ServiceProvider

22from flipdare.firestore.context.dare_context import DareContext

23from flipdare.firestore.context.friend_context import FriendContext

24from flipdare.firestore.context.group_member_context import GroupMemberContext

25from flipdare.generated import AppErrorCode, AppJobType, ModelObjType, NotificationType

26from flipdare.generated.model.internal.image_model import ImageModel

27from flipdare.generated.model.notification_model import NotificationModel

28from flipdare.generated.shared.firestore_collections import FirestoreCollections

29from flipdare.wrapper import NotificationWrapper, PersistedGuard

31if TYPE_CHECKING:

32 from flipdare.manager.db_manager import DbManager

33 from flipdare.manager.backend_manager import BackendManager

35__all__ = ["NotificationService"]

38_MSG = NotificationMessage

41class NotificationService(ServiceProvider):

42 """

43 NOTE: NotificationAdmin is only called by other admin classes, so it should

44 NOTE: only return AppResults, so the caller can then call

45 NOTE: the log_execution decorator.

46 """

48 def __init__(

49 self,

50 db_manager: DbManager | None = None,

51 backend_manager: BackendManager | None = None,

52 ) -> None:

53 super().__init__(

54 backend_manager=backend_manager,

55 db_manager=db_manager,

56 )

58 def _to_log_result(

59 self,

60 result: AppResult[NotificationWrapper],

61 doc_id: str | None,

62 ) -> JobResult[NotificationWrapper]:

63 """Convert AppResult to LogResult."""

64 if doc_id is None:

65 doc_id = result.doc_id or NO_DOC_ID

67 if result.is_error:

68 return JobResult.from_result(result, doc_id=doc_id)

69 return JobResult.ok(doc_id=doc_id)

71 @trigger_decorator(job_type=AppJobType.TR_FRIEND, collection=FirestoreCollections.FRIEND)

72 def send_friend_notif(

73 self,

74 friend_context: FriendContext,

75 is_update: bool,

76 ) -> JobResult[NotificationWrapper]:

77 friend = friend_context.friend

78 friend_id = friend.doc_id

79 assert friend_id is not None # NOTE: type narrowing

81 if is_update:

82 notif_type = NotificationType.UPDATE

83 msg = _MSG.FRIEND_REQ_ACCEPTED.format(name=friend_context.to_user.display_name)

84 else:

85 notif_type = NotificationType.REQUEST

86 msg = _MSG.FRIEND_REQ_SENT.format(name=friend_context.from_user.display_name)

88 result = self._add_user_notif(

89 from_uid=friend_context.from_user.doc_id,

90 to_uid=friend_context.to_user.doc_id,

91 notif_type=notif_type,

92 message=msg,

93 )

94 return self._to_log_result(result, friend_id)

96 @trigger_decorator(job_type=AppJobType.TR_GROUP, collection=FirestoreCollections.GROUP)

97 def send_group_notif(

98 self,

99 group_context: GroupMemberContext,

100 is_request: bool,

101 ) -> JobResult[NotificationWrapper]:

102 group = group_context.group

103 group_id = group.doc_id

104 owner_uid = group_context.owner.doc_id

105

106 if is_request:

107 notif_type = NotificationType.REQUEST

108 msg = _MSG.GROUP_REQ.format(

109 name=group_context.owner.display_name,

110 group_name=group.name,

111 )

112 else: # update/response

113 notif_type = NotificationType.UPDATE

114 msg = _MSG.GROUP_UPDATE.format(group_name=group.name)

115

116 result = self._add_group_notif(

117 group_id=group_id,

118 from_uid=owner_uid,

119 notif_type=notif_type,

120 message=msg,

121 )

122 return self._to_log_result(result, group_id)

123

124 @trigger_decorator(job_type=AppJobType.TR_DARE, collection=FirestoreCollections.DARE)

125 def send_dare_notif(

126 self,

127 dare_context: DareContext,

128 is_update: bool,

129 ) -> JobResult[NotificationWrapper]:

130 dare = dare_context.dare

131 dare_id = dare.doc_id

132

133 from_user = dare_context.from_user

134 from_uid = from_user.doc_id

135

136 notif_type: NotificationType | None = None

137 notif_msg: str | None = None

138

139 main_result = AppResult[NotificationWrapper](doc_id=dare_id)

140 if dare.is_group_dare:

141 group = dare_context.to_obj

142 if not PersistedGuard.is_group(group):

143 msg = f"Dare ({dare.doc_id}) is marked as group dare but to_obj is not GroupModel."

144 LOG().error(msg)

145 main_result.add_error(AppErrorCode.UNEXPECTED_CODE_PATH, msg)

146 else:

147 to_id = group.doc_id

148

149 if is_update:

150 notif_type = NotificationType.UPDATE

151 notif_msg = _MSG.GROUP_DARE_UPDATE.format(group_name=group.name)

152 else:

153 notif_type = NotificationType.REQUEST

154 notif_msg = _MSG.GROUP_DARE_NEW.format(group_name=group.name)

155

156 result = self._add_group_dare_notif(

157 group_id=to_id,

158 dare_id=dare_id,

159 from_uid=from_uid,

160 notif_type=notif_type,

161 message=notif_msg,

162 )

163 if main_result.is_error:

164 main_result.merge(result)

165 else:

166 user = dare_context.to_obj

167 if not PersistedGuard.is_user(user):

168 msg = f"Dare ({dare.doc_id}) is marked as user dare but to_obj is not UserWrapper."

169 LOG().error(msg)

170 main_result.add_error(AppErrorCode.UNEXPECTED_CODE_PATH, msg)

171 else:

172 to_id = user.doc_id

173 from_contact_name = from_user.model.contact_name

174 to_contact_name = user.model.contact_name

175

176 if is_update:

177 notif_type = NotificationType.UPDATE

178 notif_msg = _MSG.DARE_UPDATE.format(

179 from_name=from_contact_name,

180 to_name=to_contact_name,

181 )

182 else:

183 notif_type = NotificationType.REQUEST

184 notif_msg = _MSG.DARE_NEW.format(from_name=from_contact_name)

185

186 result = self._add_dare_notif(

187 dare_id=dare_id,

188 from_uid=from_uid,

189 to_uid=to_id,

190 notif_type=notif_type,

191 message=notif_msg,

192 )

193 if main_result.is_error:

194 main_result.merge(result)

195

196 return self._to_log_result(main_result, dare_id)

197

198 def _add_user_notif(

199 self,

200 from_uid: str,

201 to_uid: str,

202 notif_type: NotificationType,

203 message: str,

204 image_model: ImageModel | None = None,

205 ) -> AppResult[NotificationWrapper]:

207 # | -------------------- | --------------------- | ------------------- | ----------------- | ------------------ | ------------------- |

212 # | System | `N/A` use other `objType` | `Notif.system` | `ObjType.user` | `N/A` | `N/A` | `N/A` |

213

214 main_result = AppResult[NotificationWrapper](

215 doc_id=to_uid,

216 task_name=f" from {from_uid} to {to_uid}",

217 )

218

219 if notif_type == NotificationType.SYSTEM:

220 msg = (

221 f"System notifications cannot be created via user notifications "

222 f"for to_uid={to_uid}, from_uid={from_uid}"

223 )

224 LOG().error(msg)

225 main_result.add_error(AppErrorCode.UNEXPECTED_CODE_PATH, msg)

226 return main_result

227

228 notif = NotificationModel(

229 id=None,

230 notif_type=notif_type,

231 obj_id=from_uid,

232 obj_type=ModelObjType.USER,

233 to_uid=to_uid,

234 message=message,

235 image=image_model,

236 )

237

238 return self._add_notif(to_uid, notif)

239

240 def _add_dare_notif(

241 self,

242 dare_id: str,

243 from_uid: str,

244 to_uid: str,

245 notif_type: NotificationType,

246 message: str,

247 reporter_uid: str | None = None,

248 ) -> AppResult[NotificationWrapper]:

250 # | -------------------- | ------------------------- | ----------------- | ------------------- | ----------------- | ------------------ | -------------- |

256

257 actual_from_uid: str

258 main_result = AppResult[NotificationWrapper](

259 doc_id=dare_id,

260 task_name=f" from {from_uid} to {to_uid}",

261 )

262

263 if notif_type != NotificationType.SYSTEM:

264 actual_from_uid = from_uid

265 elif reporter_uid is not None:

266 actual_from_uid = reporter_uid

267 else:

268 msg = f"Reporter_id must be provided for SYSTEM notifications for dare_id={dare_id}"

269 LOG().error(msg)

270 main_result.add_error(

271 AppErrorCode.UNEXPECTED_CODE_PATH,

272 msg,

273 extra={"dare_id": dare_id, "from_uid": from_uid, "to_uid": to_uid},

274 )

275 return main_result

276

277 notif = NotificationModel(

278 id=None,

279 notif_type=notif_type,

280 from_uid=actual_from_uid,

281 obj_id=dare_id,

282 obj_type=ModelObjType.GROUP_DARE,

283 to_uid=to_uid,

284 message=message,

285 )

286 return self._add_notif(to_uid, notif)

287

288 def _add_group_dare_notif(

289 self,

290 group_id: str,

291 dare_id: str,

292 from_uid: str,

293 notif_type: NotificationType,

294 message: str,

295 to_uid: str | None = None,

296 ) -> AppResult[NotificationWrapper]:

298 # | -------------------- | ------------------------- | ----------------- | ------------------- | ----------------- | ------------------ | -------------- |

304

305 main_result = AppResult[NotificationWrapper](

306 doc_id=dare_id,

307 task_name=f" from {from_uid} to {to_uid}",

308 )

309

310 if notif_type != NotificationType.RECOMMEND:

311 return self._add_group_member_notifs(

312 group_id=group_id,

313 notif_type=notif_type,

314 obj_id=dare_id,

315 obj_type=ModelObjType.GROUP_DARE,

316 from_uid=from_uid,

317 message=message,

318 )

319 # if notif_type == NotificationType.RECOMMEND:

320 if to_uid is None:

321 msg = (

322 "to_uid must be provided for recommend notifications for "

323 f"group_id={group_id}, from_uid={from_uid}"

324 )

325 LOG().error(msg)

326 main_result.add_error(AppErrorCode.UNEXPECTED_CODE_PATH, msg)

327 return main_result

328

329 notif = NotificationModel(

330 id=None,

331 notif_type=notif_type,

332 from_uid=from_uid,

333 obj_id=group_id,

334 obj_type=ModelObjType.GROUP_DARE,

335 to_uid=to_uid,

336 message=message,

337 )

338 return self._add_notif(to_uid, notif)

339

340 def _add_group_notif(

341 self,

342 group_id: str,

343 from_uid: str,

344 notif_type: NotificationType,

345 message: str,

346 to_uid: str | None = None,

347 ) -> AppResult[NotificationWrapper]:

349 # | -------------------- | ------------------------- | ----------------- | ------------------- | ----------------- | ---------------- | -------------- |

355 main_result = AppResult[NotificationWrapper](

356 doc_id=group_id,

357 task_name=f" from {from_uid} to {to_uid}",

358 )

359

360 if notif_type != NotificationType.RECOMMEND:

361 return self._add_group_member_notifs(

362 group_id=group_id,

363 notif_type=notif_type,

364 obj_id=group_id,

365 obj_type=ModelObjType.GROUP,

366 from_uid=from_uid,

367 message=message,

368 )

369 # if notif_type == NotificationType.RECOMMEND:

370 if to_uid is None:

371 msg = (

372 "to_uid must be provided for recommend notifications for "

373 f"group_id={group_id}, from_uid={from_uid}"

374 )

375 LOG().error(msg)

376 main_result.add_error(AppErrorCode.UNEXPECTED_CODE_PATH, msg)

377 return main_result

378

379 notif = NotificationModel(

380 id=None,

381 notif_type=notif_type,

382 from_uid=from_uid,

383 obj_id=group_id,

384 obj_type=ModelObjType.GROUP,

385 to_uid=to_uid,

386 message=message,

387 )

388 return self._add_notif(to_uid, notif)

389

390 def _add_group_member_notifs(

391 self,

392 group_id: str,

393 notif_type: NotificationType,

394 obj_id: str,

395 obj_type: ModelObjType,

396 from_uid: str,

397 message: str,

398 ) -> AppResult[NotificationWrapper]:

399 main_result = AppResult[NotificationWrapper](

400 doc_id=group_id,

401 task_name=f" from {from_uid} to {obj_id}",

402 )

403

404 try:

405 group_members = self.group_db.get_members(group_id)

406 if not group_members or len(group_members) == 0:

407 msg = f"Unable to find any members of group {group_id}, can't add {notif_type}"

408 LOG().error(msg)

409 main_result.add_warning(msg)

410 return main_result

411

412 for member in group_members:

413 member_uid = member.uid

414 if IS_DEBUG:

415 LOG().debug(f"Adding notification for group member: {member_uid}")

416

417 notif = NotificationModel(

418 id=None,

419 notif_type=notif_type,

420 from_uid=from_uid,

421 obj_id=obj_id,

422 obj_type=obj_type,

423 to_uid=member_uid,

424 message=message,

425 )

426

427 notif_result = self._add_notif(member_uid, notif)

428 if notif_result.is_error:

429 main_result.merge(notif_result)

430 elif notif_result.is_skipped:

431 msg = f"Notification for group member {member_uid} was skipped."

432 LOG().warning(msg)

433 main_result.add_warning(msg)

434

435 return main_result

436

437 except Exception as error:

438 msg = f"Error adding group notifications for {group_id}: {error}"

439 LOG().error(msg)

440 main_result.add_error(AppErrorCode.SERVER_EX, msg)

441 return main_result

442

443 def _add_notif(

444 self,

445 to_user_id: str,

446 notif: NotificationModel,

447 ) -> AppResult[NotificationWrapper]:

448 result: AppResult[NotificationWrapper] = AppResult(

449 doc_id=notif.obj_id,

450 task_name=f" from {notif.from_uid} to {to_user_id}",

451 )

452

453 try:

454 LOG().info(f"Adding notification for {to_user_id}: {notif!s}")

455 if self.user_db.notif_exists(to_user_id, notif.obj_id, notif.notif_type):

456 msg = (

457 f"Notification of type {notif.notif_type} for dare {notif.obj_id} "

458 f"already exists for user {to_user_id}. Skipping creation."

459 )

460 LOG().debug(msg)

461 result.set_skipped(message=msg)

462 return result

463

464 saved_notif = self.user_db.create_notif(to_user_id, notif)

465 result.generated = saved_notif

466 return result

467 except Exception as error:

468 cause = (

469 f"Unable to add notification {notif.obj_type} for {notif.obj_id}:"

470 f" {error}\n\t{notif!s}"

471 )

472 LOG().error(cause)

473 result.add_error(AppErrorCode.DATABASE_EX, cause)

474 return result

Coverage for functions \ flipdare \ service \ notification_service.py: 93%

180 statements