def bulk_remove_subscriptions(users: Iterable[UserProfile], streams: Iterable[Stream], acting_client: Client, acting_user: Optional[UserProfile]=None) -> SubAndRemovedT:
    users = list(users)
    streams = list(streams)
    stream_dict = {stream.id: stream for stream in streams}
    existing_subs_by_user = get_bulk_stream_subscriber_info(users, stream_dict)

    def get_non_subscribed_tups() -> List[Tuple[(UserProfile, Stream)]]:
        stream_ids = {stream.id for stream in streams}
        not_subscribed: List[Tuple[(UserProfile, Stream)]] = []
        for user_profile in users:
            user_sub_stream_info = existing_subs_by_user[user_profile.id]
            subscribed_stream_ids = {stream.id for (sub, stream) in user_sub_stream_info}
            not_subscribed_stream_ids = (stream_ids - subscribed_stream_ids)
            for stream_id in not_subscribed_stream_ids:
                stream = stream_dict[stream_id]
                not_subscribed.append((user_profile, stream))
        return not_subscribed
    not_subscribed = get_non_subscribed_tups()
    subs_to_deactivate: List[Tuple[(Subscription, Stream)]] = []
    sub_ids_to_deactivate: List[int] = []
    for tup_list in existing_subs_by_user.values():
        for (sub, stream) in tup_list:
            subs_to_deactivate.append((sub, stream))
            sub_ids_to_deactivate.append(sub.id)
    our_realm = users[0].realm
    with transaction.atomic():
        occupied_streams_before = list(get_occupied_streams(our_realm))
        Subscription.objects.filter(id__in=sub_ids_to_deactivate).update(active=False)
        occupied_streams_after = list(get_occupied_streams(our_realm))
    event_time = timezone_now()
    event_last_message_id = get_last_message_id()
    all_subscription_logs: List[RealmAuditLog] = []
    for (sub, stream) in subs_to_deactivate:
        all_subscription_logs.append(RealmAuditLog(realm=sub.user_profile.realm, acting_user=acting_user, modified_user=sub.user_profile, modified_stream=stream, event_last_message_id=event_last_message_id, event_type=RealmAuditLog.SUBSCRIPTION_DEACTIVATED, event_time=event_time))
    RealmAuditLog.objects.bulk_create(all_subscription_logs)
    altered_user_dict: Dict[(int, List[UserProfile])] = defaultdict(list)
    streams_by_user: Dict[(int, List[Stream])] = defaultdict(list)
    for (sub, stream) in subs_to_deactivate:
        streams_by_user[sub.user_profile_id].append(stream)
        altered_user_dict[stream.id].append(sub.user_profile)
    for user_profile in users:
        if (len(streams_by_user[user_profile.id]) == 0):
            continue
        notify_subscriptions_removed(user_profile, streams_by_user[user_profile.id])
        event = {'type': 'mark_stream_messages_as_read', 'client_id': acting_client.id, 'user_profile_id': user_profile.id, 'stream_ids': [stream.id for stream in streams]}
        queue_json_publish('deferred_work', event)
    all_subscribers_by_stream = get_user_ids_for_streams(streams=streams)

    def send_peer_remove_event(stream: Stream) -> None:
        if (stream.is_in_zephyr_realm and (not stream.invite_only)):
            return
        altered_users = altered_user_dict[stream.id]
        altered_user_ids = [u.id for u in altered_users]
        subscribed_user_ids = all_subscribers_by_stream[stream.id]
        peer_user_ids = get_peer_user_ids_for_stream_change(stream=stream, altered_user_ids=altered_user_ids, subscribed_user_ids=subscribed_user_ids)
        if peer_user_ids:
            for removed_user in altered_users:
                event = dict(type='subscription', op='peer_remove', stream_id=stream.id, user_id=removed_user.id)
                send_event(our_realm, event, peer_user_ids)
    for stream in streams:
        send_peer_remove_event(stream=stream)
    new_vacant_streams = [stream for stream in (set(occupied_streams_before) - set(occupied_streams_after))]
    new_vacant_private_streams = [stream for stream in new_vacant_streams if stream.invite_only]
    new_vacant_public_streams = [stream for stream in new_vacant_streams if (not stream.invite_only)]
    if new_vacant_public_streams:
        event = dict(type='stream', op='vacate', streams=[stream.to_dict() for stream in new_vacant_public_streams])
        send_event(our_realm, event, active_user_ids(our_realm.id))
    if new_vacant_private_streams:
        for stream in new_vacant_private_streams:
            do_deactivate_stream(stream, acting_user=acting_user)
    return ([(sub.user_profile, stream) for (sub, stream) in subs_to_deactivate], not_subscribed)
def bulk_remove_subscriptions(users: Iterable[UserProfile], streams: Iterable[Stream], acting_client: Client, acting_user: Optional[UserProfile]=None) -> SubAndRemovedT:
    users = list(users)
    streams = list(streams)
    stream_dict = {stream.id: stream for stream in streams}
    existing_subs_by_user = get_bulk_stream_subscriber_info(users, stream_dict)

    def get_non_subscribed_tups() -> List[Tuple[(UserProfile, Stream)]]:
        stream_ids = {stream.id for stream in streams}
        not_subscribed: List[Tuple[(UserProfile, Stream)]] = []
        for user_profile in users:
            user_sub_stream_info = existing_subs_by_user[user_profile.id]
            subscribed_stream_ids = {stream.id for (sub, stream) in user_sub_stream_info}
            not_subscribed_stream_ids = (stream_ids - subscribed_stream_ids)
            for stream_id in not_subscribed_stream_ids:
                stream = stream_dict[stream_id]
                not_subscribed.append((user_profile, stream))
        return not_subscribed
    not_subscribed = get_non_subscribed_tups()
    subs_to_deactivate: List[Tuple[(Subscription, Stream)]] = []
    sub_ids_to_deactivate: List[int] = []
    for tup_list in existing_subs_by_user.values():
        for (sub, stream) in tup_list:
            subs_to_deactivate.append((sub, stream))
            sub_ids_to_deactivate.append(sub.id)
    our_realm = users[0].realm
    with transaction.atomic():
        occupied_streams_before = list(get_occupied_streams(our_realm))
        Subscription.objects.filter(id__in=sub_ids_to_deactivate).update(active=False)
        occupied_streams_after = list(get_occupied_streams(our_realm))
    event_time = timezone_now()
    event_last_message_id = get_last_message_id()
    all_subscription_logs: List[RealmAuditLog] = []
    for (sub, stream) in subs_to_deactivate:
        all_subscription_logs.append(RealmAuditLog(realm=sub.user_profile.realm, acting_user=acting_user, modified_user=sub.user_profile, modified_stream=stream, event_last_message_id=event_last_message_id, event_type=RealmAuditLog.SUBSCRIPTION_DEACTIVATED, event_time=event_time))
    RealmAuditLog.objects.bulk_create(all_subscription_logs)
    altered_user_dict: Dict[(int, List[UserProfile])] = defaultdict(list)
    streams_by_user: Dict[(int, List[Stream])] = defaultdict(list)
    for (sub, stream) in subs_to_deactivate:
        streams_by_user[sub.user_profile_id].append(stream)
        altered_user_dict[stream.id].append(sub.user_profile)
    for user_profile in users:
        if (len(streams_by_user[user_profile.id]) == 0):
            continue
        notify_subscriptions_removed(user_profile, streams_by_user[user_profile.id])
        event = {'type': 'mark_stream_messages_as_read', 'client_id': acting_client.id, 'user_profile_id': user_profile.id, 'stream_ids': [stream.id for stream in streams]}
        queue_json_publish('deferred_work', event)
    all_subscribers_by_stream = get_user_ids_for_streams(streams=streams)

    def send_peer_remove_event(stream: Stream) -> None:
        if (stream.is_in_zephyr_realm and (not stream.invite_only)):
            return
        altered_users = altered_user_dict[stream.id]
        altered_user_ids = [u.id for u in altered_users]
        subscribed_user_ids = all_subscribers_by_stream[stream.id]
        peer_user_ids = get_peer_user_ids_for_stream_change(stream=stream, altered_user_ids=altered_user_ids, subscribed_user_ids=subscribed_user_ids)
        if peer_user_ids:
            for removed_user in altered_users:
                event = dict(type='subscription', op='peer_remove', stream_id=stream.id, user_id=removed_user.id)
                send_event(our_realm, event, peer_user_ids)
    for stream in streams:
        send_peer_remove_event(stream=stream)
    new_vacant_streams = [stream for stream in (set(occupied_streams_before) - set(occupied_streams_after))]
    new_vacant_private_streams = [stream for stream in new_vacant_streams if stream.invite_only]
    new_vacant_public_streams = [stream for stream in new_vacant_streams if (not stream.invite_only)]
    if new_vacant_public_streams:
        event = dict(type='stream', op='vacate', streams=[stream.to_dict() for stream in new_vacant_public_streams])
        send_event(our_realm, event, active_user_ids(our_realm.id))
    if new_vacant_private_streams:
        for stream in new_vacant_private_streams:
            do_deactivate_stream(stream, acting_user=acting_user)
    return ([(sub.user_profile, stream) for (sub, stream) in subs_to_deactivate], not_subscribed)