notifications: Fix leaking of private stream msgs sent before subscribing.

[shubhamdhama]

Here I've suggested some refactors, feedback about how they sound to other. I'm working on top of it to fix the issue 9834.

[timabbott]

I'd spell it by_message_id, but otherwise looks good.

[timabbott]

This else is unnecessary

[timabbott]

These look reasonable

[shubhamdhama]

@timabbott I've updated the PR up to v1, so only optimization is left.

[shubhamdhama]

build failing due to another issue, which is fixed in #10093

[timabbott]

This import to avoid a circular loop is avoidable with some refactoring. I added 6bbffe0 to do the appropriate refactor to avoid this (I think it's cleaner this way anyway).

[zulipbot]

Heads up @shubhamdhama, we just merged some commits that conflict with the changes your made in this pull request! You can review this repository's recent commits to see where the conflicts occur. Please rebase your feature branch against the upstream/master branch and resolve your pull request's merge conflicts accordingly.

[timabbott]

I merged this, after adding 6bbffe0 to avoid the circular import workaround. Thanks @shubhamdhama!

For working on the optimization piece, I think we can just add a wrapper for bulk_access_messages, with a name like bulk_access_messages_in_stream, and have it pass in the Stream object (and maybe also the subscription object?), using e.g. this approach:

@@ -497,13 +497,15 @@ def access_message(user_profile: UserProfile, message_id: int) -> Tuple[Message,
     raise JsonableError(_("Invalid message(s)"))
 
 def has_message_access(user_profile: UserProfile, message: Message,
-                       user_message: Optional[UserMessage]) -> bool:
+                       user_message: Optional[UserMessage],
+                       stream: Optional[Stream]=None) -> bool:
     if user_message is None:
         if message.recipient.type != Recipient.STREAM:
             # You can't access private messages you didn't receive
             return False
 
-        stream = Stream.objects.get(id=message.recipient.type_id)
+        if stream is None:
+            stream = Stream.objects.get(id=message.recipient.type_id)
         if stream.realm != user_profile.realm:
             # You can't access public stream messages in other realms
             return False