def bulk_add_subscriptions(streams: Iterable[Stream], users: Iterable[UserProfile], color_map: Mapping[(str, str)]={}, from_stream_creation: bool=False, acting_user: Optional[UserProfile]=None) -> SubT:
    users = list(users)
    recipients_map: Dict[(int, int)] = {stream.id: stream.recipient_id for stream in streams}
    recipient_ids: List[int] = [recipient_id for recipient_id in recipients_map.values()]
    stream_map: Dict[(int, Stream)] = {}
    for stream in streams:
        stream_map[recipients_map[stream.id]] = stream
    subs_by_user: Dict[(int, List[Subscription])] = defaultdict(list)
    all_subs_query = get_stream_subscriptions_for_users(users).select_related('user_profile')
    for sub in all_subs_query:
        subs_by_user[sub.user_profile_id].append(sub)
    realm = users[0].realm
    already_subscribed: List[Tuple[(UserProfile, Stream)]] = []
    subs_to_activate: List[Tuple[(Subscription, Stream)]] = []
    new_subs: List[Tuple[(UserProfile, int, Stream)]] = []
    for user_profile in users:
        needs_new_sub: Set[int] = set(recipient_ids)
        for sub in subs_by_user[user_profile.id]:
            if (sub.recipient_id in needs_new_sub):
                needs_new_sub.remove(sub.recipient_id)
                if sub.active:
                    already_subscribed.append((user_profile, stream_map[sub.recipient_id]))
                else:
                    subs_to_activate.append((sub, stream_map[sub.recipient_id]))
                    sub.active = True
        for recipient_id in needs_new_sub:
            new_subs.append((user_profile, recipient_id, stream_map[recipient_id]))
    subs_to_add: List[Tuple[(Subscription, Stream)]] = []
    for (user_profile, recipient_id, stream) in new_subs:
        if (stream.name in color_map):
            color = color_map[stream.name]
        else:
            color = pick_color(user_profile, subs_by_user[user_profile.id])
        sub_to_add = Subscription(user_profile=user_profile, active=True, color=color, recipient_id=recipient_id)
        subs_by_user[user_profile.id].append(sub_to_add)
        subs_to_add.append((sub_to_add, stream))
    with transaction.atomic():
        occupied_streams_before = list(get_occupied_streams(realm))
        Subscription.objects.bulk_create([sub for (sub, stream) in subs_to_add])
        sub_ids = [sub.id for (sub, stream) in subs_to_activate]
        Subscription.objects.filter(id__in=sub_ids).update(active=True)
        occupied_streams_after = list(get_occupied_streams(realm))
    event_time = timezone_now()
    event_last_message_id = get_last_message_id()
    all_subscription_logs: List[RealmAuditLog] = []
    for (sub, stream) in subs_to_add:
        all_subscription_logs.append(RealmAuditLog(realm=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_CREATED, event_time=event_time))
    for (sub, stream) in subs_to_activate:
        all_subscription_logs.append(RealmAuditLog(realm=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_ACTIVATED, event_time=event_time))
    RealmAuditLog.objects.bulk_create(all_subscription_logs)
    new_occupied_streams = [stream for stream in (set(occupied_streams_after) - set(occupied_streams_before)) if (not stream.invite_only)]
    if (new_occupied_streams and (not from_stream_creation)):
        event: Dict[(str, object)] = dict(type='stream', op='occupy', streams=[stream.to_dict() for stream in new_occupied_streams])
        send_event(realm, event, active_user_ids(realm.id))
    all_subscribers_by_stream = get_user_ids_for_streams(streams=streams)

    def fetch_stream_subscriber_user_ids(stream: Stream) -> List[int]:
        if (stream.is_in_zephyr_realm and (not stream.invite_only)):
            return []
        user_ids = all_subscribers_by_stream[stream.id]
        return user_ids
    sub_tuples_by_user: Dict[(int, List[Tuple[(Subscription, Stream)]])] = defaultdict(list)
    new_streams: Set[Tuple[(int, int)]] = set()
    for (sub, stream) in (subs_to_add + subs_to_activate):
        sub_tuples_by_user[sub.user_profile.id].append((sub, stream))
        new_streams.add((sub.user_profile.id, stream.id))
    for stream in streams:
        if (not stream.is_public()):
            realm_admin_ids = [user.id for user in realm.get_admin_users_and_bots()]
            new_users_ids = [user.id for user in users if (((user.id, stream.id) in new_streams) and (user.id not in realm_admin_ids))]
            send_stream_creation_event(stream, new_users_ids)
    stream_ids = {stream.id for stream in streams}
    recent_traffic = get_streams_traffic(stream_ids=stream_ids)
    for user_profile in users:
        if (len(sub_tuples_by_user[user_profile.id]) == 0):
            continue
        sub_pairs = sub_tuples_by_user[user_profile.id]
        notify_subscriptions_added(user_profile, sub_pairs, fetch_stream_subscriber_user_ids, recent_traffic)
    for stream in streams:
        if (stream.is_in_zephyr_realm and (not stream.invite_only)):
            continue
        new_user_ids = [user.id for user in users if ((user.id, stream.id) in new_streams)]
        subscribed_user_ids = all_subscribers_by_stream[stream.id]
        peer_user_ids = get_peer_user_ids_for_stream_change(stream=stream, altered_user_ids=new_user_ids, subscribed_user_ids=subscribed_user_ids)
        if peer_user_ids:
            for new_user_id in new_user_ids:
                event = dict(type='subscription', op='peer_add', stream_id=stream.id, user_id=new_user_id)
                send_event(realm, event, peer_user_ids)
    return (([(user_profile, stream) for (user_profile, recipient_id, stream) in new_subs] + [(sub.user_profile, stream) for (sub, stream) in subs_to_activate]), already_subscribed)
def bulk_add_subscriptions(streams: Iterable[Stream], users: Iterable[UserProfile], color_map: Mapping[(str, str)]={}, from_stream_creation: bool=False, acting_user: Optional[UserProfile]=None) -> SubT:
    users = list(users)
    recipients_map: Dict[(int, int)] = {stream.id: stream.recipient_id for stream in streams}
    recipient_ids: List[int] = [recipient_id for recipient_id in recipients_map.values()]
    stream_map: Dict[(int, Stream)] = {}
    for stream in streams:
        stream_map[recipients_map[stream.id]] = stream
    subs_by_user: Dict[(int, List[Subscription])] = defaultdict(list)
    all_subs_query = get_stream_subscriptions_for_users(users).select_related('user_profile')
    for sub in all_subs_query:
        subs_by_user[sub.user_profile_id].append(sub)
    realm = users[0].realm
    already_subscribed: List[Tuple[(UserProfile, Stream)]] = []
    subs_to_activate: List[Tuple[(Subscription, Stream)]] = []
    new_subs: List[Tuple[(UserProfile, int, Stream)]] = []
    for user_profile in users:
        needs_new_sub: Set[int] = set(recipient_ids)
        for sub in subs_by_user[user_profile.id]:
            if (sub.recipient_id in needs_new_sub):
                needs_new_sub.remove(sub.recipient_id)
                if sub.active:
                    already_subscribed.append((user_profile, stream_map[sub.recipient_id]))
                else:
                    subs_to_activate.append((sub, stream_map[sub.recipient_id]))
                    sub.active = True
        for recipient_id in needs_new_sub:
            new_subs.append((user_profile, recipient_id, stream_map[recipient_id]))
    subs_to_add: List[Tuple[(Subscription, Stream)]] = []
    for (user_profile, recipient_id, stream) in new_subs:
        if (stream.name in color_map):
            color = color_map[stream.name]
        else:
            color = pick_color(user_profile, subs_by_user[user_profile.id])
        sub_to_add = Subscription(user_profile=user_profile, active=True, color=color, recipient_id=recipient_id)
        subs_by_user[user_profile.id].append(sub_to_add)
        subs_to_add.append((sub_to_add, stream))
    with transaction.atomic():
        occupied_streams_before = list(get_occupied_streams(realm))
        Subscription.objects.bulk_create([sub for (sub, stream) in subs_to_add])
        sub_ids = [sub.id for (sub, stream) in subs_to_activate]
        Subscription.objects.filter(id__in=sub_ids).update(active=True)
        occupied_streams_after = list(get_occupied_streams(realm))
    event_time = timezone_now()
    event_last_message_id = get_last_message_id()
    all_subscription_logs: List[RealmAuditLog] = []
    for (sub, stream) in subs_to_add:
        all_subscription_logs.append(RealmAuditLog(realm=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_CREATED, event_time=event_time))
    for (sub, stream) in subs_to_activate:
        all_subscription_logs.append(RealmAuditLog(realm=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_ACTIVATED, event_time=event_time))
    RealmAuditLog.objects.bulk_create(all_subscription_logs)
    new_occupied_streams = [stream for stream in (set(occupied_streams_after) - set(occupied_streams_before)) if (not stream.invite_only)]
    if (new_occupied_streams and (not from_stream_creation)):
        event: Dict[(str, object)] = dict(type='stream', op='occupy', streams=[stream.to_dict() for stream in new_occupied_streams])
        send_event(realm, event, active_user_ids(realm.id))
    all_subscribers_by_stream = get_user_ids_for_streams(streams=streams)

    def fetch_stream_subscriber_user_ids(stream: Stream) -> List[int]:
        if (stream.is_in_zephyr_realm and (not stream.invite_only)):
            return []
        user_ids = all_subscribers_by_stream[stream.id]
        return user_ids
    sub_tuples_by_user: Dict[(int, List[Tuple[(Subscription, Stream)]])] = defaultdict(list)
    new_streams: Set[Tuple[(int, int)]] = set()
    for (sub, stream) in (subs_to_add + subs_to_activate):
        sub_tuples_by_user[sub.user_profile.id].append((sub, stream))
        new_streams.add((sub.user_profile.id, stream.id))
    for stream in streams:
        if (not stream.is_public()):
            realm_admin_ids = [user.id for user in realm.get_admin_users_and_bots()]
            new_users_ids = [user.id for user in users if (((user.id, stream.id) in new_streams) and (user.id not in realm_admin_ids))]
            send_stream_creation_event(stream, new_users_ids)
    stream_ids = {stream.id for stream in streams}
    recent_traffic = get_streams_traffic(stream_ids=stream_ids)
    for user_profile in users:
        if (len(sub_tuples_by_user[user_profile.id]) == 0):
            continue
        sub_pairs = sub_tuples_by_user[user_profile.id]
        notify_subscriptions_added(user_profile, sub_pairs, fetch_stream_subscriber_user_ids, recent_traffic)
    for stream in streams:
        if (stream.is_in_zephyr_realm and (not stream.invite_only)):
            continue
        new_user_ids = [user.id for user in users if ((user.id, stream.id) in new_streams)]
        subscribed_user_ids = all_subscribers_by_stream[stream.id]
        peer_user_ids = get_peer_user_ids_for_stream_change(stream=stream, altered_user_ids=new_user_ids, subscribed_user_ids=subscribed_user_ids)
        if peer_user_ids:
            for new_user_id in new_user_ids:
                event = dict(type='subscription', op='peer_add', stream_id=stream.id, user_id=new_user_id)
                send_event(realm, event, peer_user_ids)
    return (([(user_profile, stream) for (user_profile, recipient_id, stream) in new_subs] + [(sub.user_profile, stream) for (sub, stream) in subs_to_activate]), already_subscribed)