Coverage for src/local_deep_research/notifications/manager.py: 98%

1"""

2High-level notification manager with database integration.

3"""

5from dataclasses import dataclass

6from datetime import datetime, timezone, timedelta

7from enum import Enum

8from typing import Dict, Optional, Any

9from collections import deque

10import threading

12from loguru import logger

14from ..settings.env_registry import get_env_setting

15from .service import NotificationService

16from .templates import EventType

17from .exceptions import RateLimitError

18from ..utilities.type_utils import unwrap_setting

19from ..constants import DEFAULT_SEARCH_TOOL

22class NotificationReason(str, Enum):

23 """Precise reason a notification was (not) sent.

25 Surfaced in :class:`NotificationResult` so callers and log lines can

26 distinguish, for example, a per-user toggle being off from the

27 server-level master switch being unset — see issue #4877.

29 Note: rate-limit refusal is signalled via :class:`RateLimitError`

30 (preserved API for callers that catch it directly) rather than as a

31 ``NotificationResult`` value.

32 """

34 SENT = "sent"

35 SERVER_DISABLED = "server_disabled"

36 EVENT_DISABLED = "event_disabled"

37 UNCONFIGURED = "unconfigured"

38 EGRESS_DENIED = "egress_denied"

39 WEBHOOK_FAILED = "webhook_failed"

40 EXCEPTION = "exception"

43@dataclass(frozen=True)

44class NotificationResult:

45 """Structured outcome of :meth:`NotificationManager.send_notification`.

47 ``sent`` keeps the legacy bool contract (``if result:`` keeps working —

48 see ``__bool__``) while ``reason`` names the precise cause so the queue

49 helpers and error reporter can log a useful, non-misleading line.

50 ``detail`` is a short operator-facing explanation complementing reason.

51 """

53 sent: bool

54 reason: NotificationReason

55 detail: str = ""

57 def __bool__(self) -> bool:

58 return self.sent

61class NotificationManager:

62 """

63 High-level notification manager that uses settings snapshots for

64 thread-safe access to user settings.

66 This manager is designed to be used from background threads (e.g., queue

67 processors) by passing a settings_snapshot dictionary captured from the

68 main thread.

70 **Per-User Rate Limiting:**

71 The rate limiter is shared across ALL NotificationManager instances as a

72 singleton, but supports per-user rate limit configuration. Each user has

73 their own rate limits based on their settings, which are configured when

74 the NotificationManager is initialized with the required user_id parameter.

76 **How It Works:**

77 - The first NotificationManager instance creates the shared RateLimiter

78 with default limits

79 - Each instance configures user-specific limits by passing user_id to __init__

80 - The rate limiter maintains separate counters and limits for each user

81 - Users are completely isolated - one user's limit doesn't affect others

83 Example:

84 >>> # User A with conservative limits

85 >>> snapshot_a = {"notifications.rate_limit_per_hour": 5}

86 >>> manager_a = NotificationManager(snapshot_a, user_id="user_a")

87 >>> # ✅ user_a gets 5/hour

88 >>>

89 >>> # User B with generous limits (doesn't affect User A!)

90 >>> snapshot_b = {"notifications.rate_limit_per_hour": 20}

91 >>> manager_b = NotificationManager(snapshot_b, user_id="user_b")

92 >>> # ✅ user_b gets 20/hour, user_a still has 5/hour

93 """

95 # Shared rate limiter instance across all NotificationManager instances

96 # This ensures rate limits are enforced correctly even when multiple

97 # NotificationManager instances are created

98 _shared_rate_limiter: Optional["RateLimiter"] = None

99 _rate_limiter_lock = threading.Lock()

100

101 def __init__(self, settings_snapshot: Dict[str, Any], user_id: str):

102 """

103 Initialize the notification manager.

104

105 Args:

106 settings_snapshot: Dictionary of settings key-value pairs captured

107 from SettingsManager.get_settings_snapshot().

108 This allows thread-safe access to user settings

109 from background threads.

110 user_id: User identifier for per-user rate limiting. The rate limits

111 from settings_snapshot will be configured for this user.

112

113 Example:

114 >>> # In main thread with database session

115 >>> settings_manager = SettingsManager(session)

116 >>> snapshot = settings_manager.get_settings_snapshot()

117 >>>

118 >>> # In background thread (thread-safe)

119 >>> notification_manager = NotificationManager(

120 ... settings_snapshot=snapshot,

121 ... user_id="user123"

122 ... )

123 >>> notification_manager.send_notification(...)

124 """

125 # Store settings snapshot for thread-safe access

126 self._settings_snapshot = settings_snapshot

127 self._user_id = user_id

128

129 # Security: read from server-side environment variable only — never from

130 # user-writable DB settings. Previously this was read via

131 # _get_setting("notifications.allow_private_ips"), which allowed any

132 # user to bypass SSRF protection through the settings API.

133 # Registered as env-only in settings/env_definitions/security.py

134 # so SettingsManager will always read LDR_NOTIFICATIONS_ALLOW_PRIVATE_IPS

135 # from the environment, never from the database.

136 allow_private_ips = get_env_setting(

137 "notifications.allow_private_ips", False

138 )

139

140 # Server-level master switch (env-only). Distinct from the per-user

141 # notifications.enabled toggle in the UI: this gate is set by the

142 # deployment operator and cannot be flipped via the user-writable

143 # settings API. Disabled by default; enabling it accepts the

144 # documented DNS-rebinding TOCTOU residual risk

145 # (see SECURITY.md "Notification Webhook SSRF").

146 self._outbound_allowed = bool(

147 get_env_setting("notifications.allow_outbound", False)

148 )

149

150 self.service = NotificationService(

151 allow_private_ips=allow_private_ips,

152 outbound_allowed=self._outbound_allowed,

153 )

154

155 # Initialize shared rate limiter on first use

156 # The shared rate limiter now supports per-user limits, so each user's

157 # settings are respected regardless of initialization order.

158 with NotificationManager._rate_limiter_lock:

159 if NotificationManager._shared_rate_limiter is None:

160 # Create shared rate limiter with default limits

161 # (individual users can have different limits)

162 default_max_per_hour = self._get_setting(

163 "notifications.rate_limit_per_hour", default=10

164 )

165 default_max_per_day = self._get_setting(

166 "notifications.rate_limit_per_day", default=50

167 )

168

169 logger.info(

170 f"Initializing shared rate limiter with defaults: "

171 f"{default_max_per_hour}/hour, {default_max_per_day}/day"

172 )

173

174 NotificationManager._shared_rate_limiter = RateLimiter(

175 max_per_hour=default_max_per_hour,

176 max_per_day=default_max_per_day,

177 )

178

179 # Use the shared instance

180 self._rate_limiter = NotificationManager._shared_rate_limiter

181

182 # Configure per-user rate limits

183 max_per_hour = self._get_setting(

184 "notifications.rate_limit_per_hour", default=10

185 )

186 max_per_day = self._get_setting(

187 "notifications.rate_limit_per_day", default=50

188 )

189

190 self._rate_limiter.set_user_limits(

191 user_id, max_per_hour, max_per_day

192 )

193 logger.debug(

194 f"Configured rate limits for user {user_id}: "

195 f"{max_per_hour}/hour, {max_per_day}/day"

196 )

197

198 def _get_setting(self, key: str, default: Any = None) -> Any:

199 """

200 Get a setting value from snapshot.

201

202 Args:

203 key: Setting key

204 default: Default value if not found

205

206 Returns:

207 Setting value or default

208 """

209 return self._settings_snapshot.get(key, default)

210

211 def send_notification(

212 self,

213 event_type: EventType,

214 context: Dict[str, Any],

215 force: bool = False,

216 ) -> NotificationResult:

217 """

218 Send a notification for an event.

219

220 Uses the user_id that was provided during initialization for

221 rate limiting and user preferences.

222

223 Args:

224 event_type: Type of event

225 context: Context data for the notification

226 force: If True, bypass rate limiting

227

228 Returns:

229 :class:`NotificationResult` describing the outcome. The

230 dataclass is truthy iff the notification was actually sent,

231 so legacy ``if manager.send_notification(...):`` callers

232 continue to work without source changes.

233

234 Raises:

235 RateLimitError: If rate limit is exceeded and force=False

236 """

237 logger.debug(

238 f"Sending notification: event_type={event_type.value}, "

239 f"user_id={self._user_id}, force={force}"

240 )

241 logger.debug(f"Context keys: {list(context.keys())}")

242

243 # Server-level master switch (env-only). Disabled by default;

244 # flipping it on is the operator's acknowledgement of the

245 # documented SSRF rebinding residual risk. force=True does NOT

246 # bypass this — it bypasses the per-user event-type and

247 # rate-limit toggles only. WARNING level so an operator wondering

248 # "why aren't notifications firing?" sees the actionable signal

249 # at default log level.

250 if not self._outbound_allowed:

251 logger.warning(

252 "Notification refused: outbound notifications are disabled "

253 "at the server level. Set "

254 "LDR_NOTIFICATIONS_ALLOW_OUTBOUND=true to enable. See "

255 "SECURITY.md 'Notification Webhook SSRF' for the rationale "

256 "and residual risk. (event={}, user={})",

257 event_type.value,

258 self._user_id,

259 )

260 return NotificationResult(

261 sent=False,

262 reason=NotificationReason.SERVER_DISABLED,

263 detail="LDR_NOTIFICATIONS_ALLOW_OUTBOUND is not set",

264 )

265

266 # Check if notifications are enabled for this event type

267 should_notify = self._should_notify(event_type)

268 logger.debug(

269 f"Notification enabled check for {event_type.value}: "

270 f"{should_notify}"

271 )

272

273 if not force and not should_notify:

274 logger.debug(

275 f"Notifications disabled for event type: "

276 f"{event_type.value} (user: {self._user_id})"

277 )

278 return NotificationResult(

279 sent=False,

280 reason=NotificationReason.EVENT_DISABLED,

281 detail=f"notifications.on_{event_type.value} is off",

282 )

283

284 # Check rate limit using the manager's user_id

285 rate_limit_ok = self._rate_limiter.is_allowed(self._user_id)

286 logger.debug(f"Rate limit check for {self._user_id}: {rate_limit_ok}")

287

288 if not force and not rate_limit_ok:

289 logger.warning(f"Rate limit exceeded for user {self._user_id}")

290 raise RateLimitError(

291 "Notification rate limit exceeded. "

292 "Please wait before sending more notifications."

293 )

294

295 try:

296 # Get service URLs from settings (snapshot or database)

297 service_urls = self._get_setting(

298 "notifications.service_url", default=""

299 )

300

301 if not service_urls or not service_urls.strip():

302 logger.debug(

303 f"No notification service URLs configured for user "

304 f"{self._user_id}"

305 )

306 return NotificationResult(

307 sent=False,

308 reason=NotificationReason.UNCONFIGURED,

309 detail="notifications.service_url is not configured",

310 )

311

312 # Egress policy: gate each parsed webhook URL against the

313 # user's scope. Apprise URLs can be Slack/Discord/SMTP/HTTP —

314 # we treat anything with a host as an HTTP egress and ask

315 # evaluate_url. Schemes without a hostname (e.g. mailto)

316 # pass through; SSRF + provider-side checks still apply.

317 allowed_urls = self._filter_urls_by_egress_policy(service_urls)

318 if not allowed_urls:

319 logger.bind(policy_audit=True).warning(

320 "all notification URLs refused by egress policy",

321 user=self._user_id,

322 event=event_type.value,

323 )

324 return NotificationResult(

325 sent=False,

326 reason=NotificationReason.EGRESS_DENIED,

327 detail="all configured URLs refused by egress policy",

328 )

329

330 # Send notification with the allowed subset.

331 logger.debug(f"Calling service.send_event for {event_type.value}")

332 sent = self.service.send_event(

333 event_type, context, service_urls=allowed_urls

334 )

335

336 # Log to database if enabled

337 if sent:

338 self._log_notification(event_type, context)

339 logger.info(

340 f"Notification sent: {event_type.value} to user "

341 f"{self._user_id}"

342 )

343 return NotificationResult(

344 sent=True,

345 reason=NotificationReason.SENT,

346 detail="",

347 )

348 logger.warning(

349 f"Notification failed: {event_type.value} to user "

350 f"{self._user_id}"

351 )

352 return NotificationResult(

353 sent=False,

354 reason=NotificationReason.WEBHOOK_FAILED,

355 detail="Apprise dispatch returned False",

356 )

357

358 except Exception as exc:

359 # The exception string may contain webhook URLs with embedded

360 # tokens, SMTP credentials, or payload content. Log the raw

361 # exception for operators via logger.exception above, but only

362 # expose the exception class name (and not the message) in the

363 # structured result so downstream callers can't echo it into

364 # logs or UI surfaces verbatim.

365 logger.exception(

366 f"Error sending notification for {event_type.value} to user "

367 f"{self._user_id}"

368 )

369 return NotificationResult(

370 sent=False,

371 reason=NotificationReason.EXCEPTION,

372 detail=f"dispatch raised {type(exc).__name__}",

373 )

374

375 def test_service(self, url: str) -> Dict[str, Any]:

376 """

377 Test a notification service.

378

379 Args:

380 url: Service URL to test

381

382 Returns:

383 Dict with test results

384 """

385 # Egress policy precheck before forwarding to Apprise.

386 # /api/notifications/test-url is a user-supplied URL endpoint.

387 allowed = self._filter_urls_by_egress_policy(url)

388 if not allowed:

389 return {

390 "status": "error",

391 "message": (

392 "URL refused by egress policy. Set Egress Scope to "

393 "'Unprotected' (or pick a local webhook) to test this URL."

394 ),

395 }

396 return self.service.test_service(allowed)

397

398 def _filter_urls_by_egress_policy(self, service_urls: str) -> str:

399 """Filter an Apprise URL string (space- or comma-separated) by

400 the user's egress policy. Returns the joined string of allowed

401 URLs (may be empty). When no snapshot / context is available,

402 returns the input unchanged (back-compat — the snapshot-less

403 path was already used in older callers).

404 """

405 snapshot = getattr(self, "_settings_snapshot", None)

406 if not snapshot:

407 return service_urls

408 try:

409 from ..security.egress.policy import (

410 EgressScope,

411 PolicyDeniedError,

412 context_from_snapshot,

413 evaluate_url,

414 )

415 except ImportError:

416 logger.debug(

417 "egress_policy unavailable in notifications manager; "

418 "URLs will not be scope-gated"

419 )

420 return service_urls

421

422 primary_raw = unwrap_setting(

423 snapshot.get("search.tool", DEFAULT_SEARCH_TOOL)

424 )

425 try:

426 ctx = context_from_snapshot(

427 snapshot, primary_raw or DEFAULT_SEARCH_TOOL

428 )

429 except (PolicyDeniedError, ValueError) as exc:

430 # Snapshot present but policy cannot be evaluated. The

431 # previous bare-except dropped to "return service_urls"

432 # here, which dispatched every URL unfiltered — fail-open

433 # on a misconfigured policy. Now refuse all URLs instead

434 # so an Apprise dispatch with no scope check is impossible.

435 logger.bind(policy_audit=True).warning(

436 "all notification URLs refused: egress policy could "

437 "not be evaluated",

438 reason=str(exc),

439 )

440 return ""

441

442 # Apprise accepts space- or comma- separated URLs. Apprise has

443 # its own non-http schemes (discord://, slack://, telegram://,

444 # mailto://, msteams://, ntfy://, ...) that dispatch to EXTERNAL

445 # vendor APIs. evaluate_url only understands http/https.

446 #

447 # Under PRIVATE_ONLY ("nothing leaves the box") we cannot verify a

448 # vendor token resolves to a local endpoint, so a non-http scheme is

449 # refused — fail closed. A self-hosted notifier (local ntfy, SMTP)

450 # should be addressed by its http(s):// URL, which evaluate_url then

451 # allows as a private host. Under the other scopes the prior

452 # pass-through stands (the modeled threat there is internal-http

453 # SSRF via a raw webhook, not vendor APIs).

454 parts = []

455 for url_entry in service_urls.replace(",", " ").split():

456 url_entry = url_entry.strip()

457 if not url_entry: 457 ↛ 458line 457 didn't jump to line 458 because the condition on line 457 was never true

458 continue

459 scheme = (

460 url_entry.split(":", 1)[0].lower() if ":" in url_entry else ""

461 )

462 if scheme not in ("http", "https"):

463 if ctx.scope == EgressScope.PRIVATE_ONLY:

464 logger.bind(policy_audit=True).warning(

465 "notification refused under PRIVATE_ONLY egress "

466 "scope: non-http vendor scheme cannot be verified "

467 "local",

468 scheme=scheme or "(none)",

469 )

470 continue

471 parts.append(url_entry)

472 continue

473 decision = evaluate_url(url_entry, ctx)

474 if decision.allowed:

475 parts.append(url_entry)

476 else:

477 logger.bind(policy_audit=True).warning(

478 "notification URL refused by egress policy",

479 url=url_entry,

480 scope=ctx.scope.value,

481 reason=decision.reason,

482 )

483 return " ".join(parts)

484

485 def _should_notify(self, event_type: EventType) -> bool:

486 """

487 Check if notifications should be sent for this event type.

488

489 Uses the manager's settings snapshot to determine if the event type

490 is enabled for the user.

491

492 Args:

493 event_type: Event type to check

494

495 Returns:

496 True if notifications should be sent

497 """

498 try:

499 # Check event-specific setting (from snapshot or database)

500 setting_key = f"notifications.on_{event_type.value}"

501 enabled = self._get_setting(setting_key, default=False)

502

503 return bool(enabled)

504

505 except Exception:

506 logger.warning("Error checking notification preferences")

507 # Default to disabled on error to avoid infinite loops during login

508 return False

509

510 def _log_notification(

511 self, event_type: EventType, context: Dict[str, Any]

512 ) -> None:

513 """

514 Log a sent notification (simplified logging to application logs only).

515

516 Uses the manager's user_id for logging.

517

518 Args:

519 event_type: Event type

520 context: Notification context

521 """

522 try:

523 title = (

524 context.get("query")

525 or context.get("subscription_name")

526 or "Unknown"

527 )

528 logger.info(

529 f"Notification sent: {event_type.value} - {title} "

530 f"(user: {self._user_id})"

531 )

532 except Exception as e:

533 logger.debug(f"Failed to log notification: {e}")

534

535

536class RateLimiter:

537 """

538 Simple in-memory rate limiter for notifications with per-user limit support.

539

540 This rate limiter tracks notification counts per user and enforces

541 configurable rate limits. Each user can have their own rate limits,

542 which are stored separately from the notification counts.

543

544 **Per-User Limits:**

545 Rate limits can be configured per-user using `set_user_limits()`.

546 If no user-specific limits are set, the default limits (passed to

547 __init__) are used.

548

549 **Memory Storage:**

550 This implementation stores rate limits in memory only, which means

551 limits are reset when the server restarts. This is acceptable for normal

552 users since they cannot restart the server. If an admin restarts the server,

553 rate limits reset which is reasonable behavior.

554

555 **Thread Safety:**

556 This implementation is thread-safe using threading.Lock() for concurrent

557 requests from the same user.

558

559 **Multi-Worker Limitation:**

560 In multi-worker deployments, each worker process maintains its own rate

561 limit counters. Users could potentially bypass rate limits by distributing

562 requests across different workers, getting up to N × max_per_hour

563 notifications (where N = number of workers). For single-worker deployments

564 (the default for LDR), this is not a concern. For production multi-worker

565 deployments, consider implementing Redis-based rate limiting.

566

567 Example:

568 >>> limiter = RateLimiter(max_per_hour=10, max_per_day=50)

569 >>> # Set custom limits for specific user

570 >>> limiter.set_user_limits("user_a", max_per_hour=5, max_per_day=25)

571 >>> limiter.set_user_limits("user_b", max_per_hour=20, max_per_day=100)

572 >>> # Users get their configured limits

573 >>> limiter.is_allowed("user_a") # Limited to 5/hour

574 >>> limiter.is_allowed("user_b") # Limited to 20/hour

575 >>> limiter.is_allowed("user_c") # Uses defaults: 10/hour

576 """

577

578 def __init__(

579 self,

580 max_per_hour: int = 10,

581 max_per_day: int = 50,

582 cleanup_interval_hours: int = 24,

583 ):

584 """

585 Initialize rate limiter with default limits.

586

587 Args:

588 max_per_hour: Default maximum notifications per hour per user

589 max_per_day: Default maximum notifications per day per user

590 cleanup_interval_hours: How often to run cleanup of inactive users (hours)

591 """

592 # Default limits used when no user-specific limits are set

593 self.max_per_hour = max_per_hour

594 self.max_per_day = max_per_day

595 self.cleanup_interval_hours = cleanup_interval_hours

596

597 # Per-user rate limit configuration (user_id -> (max_per_hour, max_per_day))

598 self._user_limits: Dict[str, tuple[int, int]] = {}

599

600 # Per-user notification counts

601 self._hourly_counts: Dict[str, deque] = {}

602 self._daily_counts: Dict[str, deque] = {}

603

604 self._last_cleanup = datetime.now(timezone.utc)

605 self._lock = threading.Lock() # Thread safety for all operations

606

607 def set_user_limits(

608 self, user_id: str, max_per_hour: int, max_per_day: int

609 ) -> None:

610 """

611 Set rate limits for a specific user.

612

613 This allows each user to have their own rate limit configuration.

614 If not set, the user will use the default limits passed to __init__.

615

616 Args:

617 user_id: User identifier

618 max_per_hour: Maximum notifications per hour for this user

619 max_per_day: Maximum notifications per day for this user

620

621 Example:

622 >>> limiter = RateLimiter(max_per_hour=10, max_per_day=50)

623 >>> limiter.set_user_limits("power_user", max_per_hour=20, max_per_day=100)

624 >>> limiter.set_user_limits("limited_user", max_per_hour=5, max_per_day=25)

625 """

626 with self._lock:

627 self._user_limits[user_id] = (max_per_hour, max_per_day)

628 logger.debug(

629 f"Set rate limits for user {user_id}: "

630 f"{max_per_hour}/hour, {max_per_day}/day"

631 )

632

633 def get_user_limits(self, user_id: str) -> tuple[int, int]:

634 """

635 Get the effective rate limits for a user.

636

637 Returns the user-specific limits if set, otherwise returns defaults.

638

639 Args:

640 user_id: User identifier

641

642 Returns:

643 Tuple of (max_per_hour, max_per_day)

644 """

645 with self._lock:

646 return self._user_limits.get(

647 user_id, (self.max_per_hour, self.max_per_day)

648 )

649

650 def is_allowed(self, user_id: str) -> bool:

651 """

652 Check if a notification is allowed for a user.

653

654 Uses per-user rate limits if configured via set_user_limits(),

655 otherwise uses the default limits from __init__.

656

657 Args:

658 user_id: User identifier

659

660 Returns:

661 True if notification is allowed, False if rate limit exceeded

662 """

663 with self._lock:

664 now = datetime.now(timezone.utc)

665

666 # Periodic cleanup of inactive users

667 self._cleanup_inactive_users_if_needed(now)

668

669 # Initialize queues for user if needed

670 if user_id not in self._hourly_counts:

671 self._hourly_counts[user_id] = deque()

672 self._daily_counts[user_id] = deque()

673

674 # Clean old entries

675 self._clean_old_entries(user_id, now)

676

677 # Get user-specific limits or defaults

678 max_per_hour, max_per_day = self._user_limits.get(

679 user_id, (self.max_per_hour, self.max_per_day)

680 )

681

682 # Check limits

683 hourly_count = len(self._hourly_counts[user_id])

684 daily_count = len(self._daily_counts[user_id])

685

686 if hourly_count >= max_per_hour:

687 logger.warning(

688 f"Hourly rate limit exceeded for user {user_id}: "

689 f"{hourly_count}/{max_per_hour}"

690 )

691 return False

692

693 if daily_count >= max_per_day:

694 logger.warning(

695 f"Daily rate limit exceeded for user {user_id}: "

696 f"{daily_count}/{max_per_day}"

697 )

698 return False

699

700 # Record this notification

701 self._hourly_counts[user_id].append(now)

702 self._daily_counts[user_id].append(now)

703

704 return True

705

706 def _clean_old_entries(self, user_id: str, now: datetime) -> None:

707 """

708 Remove old entries from rate limit counters.

709

710 Args:

711 user_id: User identifier

712 now: Current time

713 """

714 hour_ago = now - timedelta(hours=1)

715 day_ago = now - timedelta(days=1)

716

717 # Clean hourly queue

718 while (

719 self._hourly_counts[user_id]

720 and self._hourly_counts[user_id][0] < hour_ago

721 ):

722 self._hourly_counts[user_id].popleft()

723

724 # Clean daily queue

725 while (

726 self._daily_counts[user_id]

727 and self._daily_counts[user_id][0] < day_ago

728 ):

729 self._daily_counts[user_id].popleft()

730

731 def reset(self, user_id: Optional[str] = None) -> None:

732 """

733 Reset rate limits for a user or all users.

734

735 Args:

736 user_id: User to reset, or None for all users

737 """

738 with self._lock:

739 if user_id:

740 self._hourly_counts.pop(user_id, None)

741 self._daily_counts.pop(user_id, None)

742 else:

743 self._hourly_counts.clear()

744 self._daily_counts.clear()

745

746 def _cleanup_inactive_users_if_needed(self, now: datetime) -> None:

747 """

748 Periodically clean up data for inactive users to prevent memory leaks.

749

750 Args:

751 now: Current timestamp

752 """

753 # Check if cleanup is needed

754 if now - self._last_cleanup < timedelta(

755 hours=self.cleanup_interval_hours

756 ):

757 return

758

759 logger.debug("Running periodic cleanup of inactive notification users")

760

761 # Define inactive threshold (users with no activity for 7 days)

762 inactive_threshold = now - timedelta(days=7)

763

764 inactive_users = []

765

766 # Find users with no recent activity

767 # Convert to list to avoid "dictionary changed size during iteration" error

768 for user_id in list(self._hourly_counts.keys()):

769 # Check if user has any recent entries

770 hourly_entries: list = list(self._hourly_counts.get(user_id, []))

771 daily_entries: list = list(self._daily_counts.get(user_id, []))

772

773 # If no entries or all entries are old, mark as inactive

774 has_recent_activity = False

775 for entry in hourly_entries + daily_entries:

776 if entry > inactive_threshold:

777 has_recent_activity = True

778 break

779

780 if not has_recent_activity:

781 inactive_users.append(user_id)

782

783 # Remove inactive users

784 for user_id in inactive_users:

785 self._hourly_counts.pop(user_id, None)

786 self._daily_counts.pop(user_id, None)

787 logger.debug(

788 f"Cleaned up inactive user {user_id} from rate limiter"

789 )

790

791 if inactive_users: 791 ↛ 796line 791 didn't jump to line 796 because the condition on line 791 was always true

792 logger.info(

793 f"Cleaned up {len(inactive_users)} inactive users from rate limiter"

794 )

795

796 self._last_cleanup = now