Merge pull request #196 from GetStream/feature/uni-91-filter-out-shadowed-messages-in-the-channel

Implement hiding shadowed messages in the IStreamChannel messages and…

Author: sierpinskid

Assets/Plugins/StreamChat/Core/Helpers/FilteredList.cs

@@ -0,0 +1,79 @@
+using System;
+using System.Collections;
+using System.Collections.Generic;
+
+namespace StreamChat.Core.Helpers
+{
+    /// <summary>
+    /// Filtered list where items can only be added if they pass the filter.
+    /// </summary>
+    /// <typeparam name="T"></typeparam>
+    internal class FilteredList<T> : IList<T>, ICollection<T>, IEnumerable<T>, IReadOnlyList<T>, IReadOnlyCollection<T>
+    {
+        public int Count => _internalList.Count;
+        public bool IsReadOnly => false;
+
+        public FilteredList(Predicate<T> addFilter)
+        {
+            _addFilter = addFilter ?? throw new ArgumentNullException(nameof(addFilter));
+        }
+
+        public IEnumerator<T> GetEnumerator() => _internalList.GetEnumerator();
+
+        public void Add(T item)
+        {
+            if (_addFilter(item))
+            {
+                _internalList.Add(item);
+            }
+            else
+            {
+                throw new InvalidOperationException("Item does not pass the filter.");
+            }
+        }
+
+        public void Clear() => _internalList.Clear();
+
+        public bool Contains(T item) => _internalList.Contains(item);
+
+        public void CopyTo(T[] array, int arrayIndex) => _internalList.CopyTo(array, arrayIndex);
+
+        public bool Remove(T item) => _internalList.Remove(item);
+
+        public int IndexOf(T item) => _internalList.IndexOf(item);
+
+        public void Insert(int index, T item)
+        {
+            if (_addFilter(item))
+            {
+                _internalList.Insert(index, item);
+            }
+            else
+            {
+                throw new InvalidOperationException("Item does not pass the filter.");
+            }
+        }
+
+        public void RemoveAt(int index) => _internalList.RemoveAt(index);
+
+        public T this[int index]
+        {
+            get => _internalList[index];
+            set
+            {
+                if (!_addFilter(value))
+                {
+                    throw new InvalidOperationException("Item does not pass the filter.");
+                }
+                _internalList[index] = value;
+            }
+        }
+
+        public void Sort(IComparer<T> comparer) => _internalList.Sort(comparer);
+
+        IEnumerator IEnumerable.GetEnumerator() => GetEnumerator();
+
+        private readonly Predicate<T> _addFilter;
+        private readonly List<T> _internalList = new List<T>();
+    }
+}

Assets/Plugins/StreamChat/Core/Helpers/FilteredList.cs.meta

@@ -28,6 +28,23 @@ public static bool ContainsNoAlloc<TItem>(this List<TItem> source, TItem item)
             return false;
         }
 
+        /// <summary>
+        /// In Unity 2019.4.40f1 List.Contains allocates memory. Use this allocation free alternative
+        /// </summary>
+        [Pure]
+        public static bool ContainsNoAlloc<TItem>(this IList<TItem> source, TItem item)
+        {
+            for (var i = 0; i < source.Count; i++)
+            {
+                if (EqualityComparer<TItem>.Default.Equals(source[i], item))
+                {
+                    return true;
+                }
+            }
+
+            return false;
+        }
+        
         [Pure]
         public static List<TDto> TrySaveToDtoCollection<TSource, TDto>(this List<TSource> source)
             where TSource : ISavableTo<TDto>

Assets/Plugins/StreamChat/Core/Helpers/ICollectionExt.cs

@@ -8,6 +8,7 @@ namespace StreamChat.Core.Helpers
     /// </summary>
     internal static class ILoadableFromExt
     {
+        //StreamTOdo: rename to TryCreateOrLoadFromDto
         /// <summary>
         /// Load domain object from the DTO. If the loadable is null, creates a new instance of the domain object.
         /// </summary>

Assets/Plugins/StreamChat/Core/Helpers/ILoadableFromExt.cs

@@ -28,15 +28,21 @@ public static void TryReplaceTrackedObjects<TTracked, TDto>(this IList<TTracked>
             foreach (var dto in dtos)
             {
                 var trackedItem = repository.CreateOrUpdate<TTracked, TDto>(dto, out _);
-                target.Add(trackedItem);
+                try
+                {
+                    target.Add(trackedItem);
+                }
+                catch
+                {
+                }
             }
         }
 
         /// <summary>
-        /// Clear target list and replace with items created or updated from DTO collection
+        /// Append target list with items created from DTO enumeration. Fails silently if target or source are null.
         /// </summary>
-        public static void TryAppendUniqueTrackedObjects<TTracked, TDto>(this IList<TTracked> target, IEnumerable<TDto> dtos,
-            ICacheRepository<TTracked> repository)
+        public static void TryAppendUniqueTrackedObjects<TTracked, TDto>(this IList<TTracked> target,
+            IEnumerable<TDto> dtos, ICacheRepository<TTracked> repository)
             where TTracked : class, IStreamStatefulModel, IUpdateableFrom<TDto, TTracked>
         {
             if (target == null)
@@ -49,6 +55,7 @@ public static void TryAppendUniqueTrackedObjects<TTracked, TDto>(this IList<TTra
                 return;
             }
 
+            //StreamTODO: change this to obtaining a hashset from a pool so (1) concurrent executions don't interfere (2) we use Hashset<TTracked> so there's no boxing
             _uniqueElements.Clear();
 
             foreach (var t in target)
@@ -62,7 +69,13 @@ public static void TryAppendUniqueTrackedObjects<TTracked, TDto>(this IList<TTra
 
                 if (_uniqueElements.Add(trackedItem))
                 {
-                    target.Add(trackedItem);
+                    try
+                    {
+                        target.Add(trackedItem);
+                    }
+                    catch
+                    {
+                    }
                 }
             }
         }

Assets/Plugins/StreamChat/Core/State/IRepositoryExt.cs

@@ -196,7 +196,7 @@ public async Task<IStreamMessage> SendNewMessageAsync(StreamSendMessageRequest s
 
             //StreamTodo: we update internal cache message without server confirmation that message got accepted. e.g. message could be rejected
             //It's ok to update the cache "in good faith" to not introduce update delay but we should handle if message got rejected
-            var streamMessage = InternalAppendOrUpdateMessage(response.Message);
+            InternalAppendOrUpdateMessage(response.Message, out var streamMessage);
             return streamMessage;
         }
 
@@ -687,7 +687,7 @@ void IUpdateableFrom<UpdateChannelResponseInternalDTO, StreamChannel>.UpdateFrom
         internal void HandleMessageNewEvent(MessageNewEventInternalDTO dto)
         {
             AssertCid(dto.Cid);
-            InternalAppendOrUpdateMessage(dto.Message);
+            InternalAppendOrUpdateMessage(dto.Message, out _);
 
             //StreamTodo: how can user react to this change? WatcherCount could internally fire WatchCountChanged event
             WatcherCount = GetOrDefault(dto.WatcherCount, WatcherCount);
@@ -696,7 +696,7 @@ internal void HandleMessageNewEvent(MessageNewEventInternalDTO dto)
         internal void InternalHandleMessageNewNotification(NotificationNewMessageEventInternalDTO dto)
         {
             AssertCid(dto.Cid);
-            InternalAppendOrUpdateMessage(dto.Message);
+            InternalAppendOrUpdateMessage(dto.Message, out _);
 
             MemberCount = dto.ChannelMemberCount;
         }
@@ -806,15 +806,28 @@ protected override string InternalUniqueId
         }
 
         private readonly List<StreamChannelMember> _members = new List<StreamChannelMember>();
-        private readonly List<StreamMessage> _messages = new List<StreamMessage>();
-        private readonly List<StreamMessage> _pinnedMessages = new List<StreamMessage>();
+        private readonly FilteredList<StreamMessage> _messages = new FilteredList<StreamMessage>(MessageFilter);
+        private readonly FilteredList<StreamMessage> _pinnedMessages = new FilteredList<StreamMessage>(MessageFilter);
         private readonly List<StreamUser> _watchers = new List<StreamUser>();
         private readonly List<StreamRead> _read = new List<StreamRead>();
         private readonly List<string> _ownCapabilities = new List<string>();
         private readonly List<StreamPendingMessage> _pendingMessages = new List<StreamPendingMessage>();
+        
+        private readonly MessageCreateAtComparer _messageCreateAtComparer = new MessageCreateAtComparer();
 
         private bool _muted;
         private bool _hidden;
+        
+        //StreamTodo: move outside and change to internal
+        private class MessageCreateAtComparer : IComparer<IStreamMessage>
+        {
+            public int Compare(IStreamMessage x, IStreamMessage y)
+            {
+                return x.CreatedAt.CompareTo(y.CreatedAt);
+            }
+        }
+
+        private static bool MessageFilter(IStreamMessage message) => !message.Shadowed.HasValue || !message.Shadowed.Value;
 
         private void AssertCid(string cid)
         {
@@ -824,42 +837,38 @@ private void AssertCid(string cid)
             }
         }
 
-        private StreamMessage InternalAppendOrUpdateMessage(MessageInternalDTO dto)
+        private bool InternalAppendOrUpdateMessage(MessageInternalDTO dto, out StreamMessage streamMessage)
         {
-            var streamMessage = Cache.TryCreateOrUpdate(dto, out var wasCreated);
+            streamMessage = Cache.TryCreateOrUpdate(dto, out var wasCreated);
             if (wasCreated)
             {
                 if (!_messages.ContainsNoAlloc(streamMessage))
                 {
                     var lastMessage = _messages.LastOrDefault();
 
-                    _messages.Add(streamMessage);
+                    try
+                    {
+                        _messages.Add(streamMessage);
+                    }
+                    catch
+                    {
+                        streamMessage = null;
+                        return false;
+                    }
 
                     // If local user sends message during the sync operation.
                     // It is possible that the locally sent message will be added before the /sync endpoint returns past message events
                     if (lastMessage != null && streamMessage.CreatedAt < lastMessage.CreatedAt)
                     {
-                        //StreamTodo: test this more. A good way was to toggle Ethernet on PC and send messages on Android
+                        //StreamTodo: test this more. One way is to toggle Ethernet on PC and send messages from Android client
                         _messages.Sort(_messageCreateAtComparer);
                     }
 
                     MessageReceived?.Invoke(this, streamMessage);
                 }
             }
 
-            return streamMessage;
-        }
-
-        //StreamTodo: move this to the right place
-        private MessageCreateAtComparer _messageCreateAtComparer = new MessageCreateAtComparer();
-
-        //StreamTodo: move outside and change to internal
-        private class MessageCreateAtComparer : IComparer<IStreamMessage>
-        {
-            public int Compare(IStreamMessage x, IStreamMessage y)
-            {
-                return x.CreatedAt.CompareTo(y.CreatedAt);
-            }
+            return true;
         }
 
         //StreamTodo: This deleteBeforeCreatedAt date is the date of event, it does not equal the passed TruncatedAt
@@ -890,7 +899,7 @@ private void InternalTruncateMessages(DateTimeOffset? deleteBeforeCreatedAt = nu
 
             if (systemMessageDto != null)
             {
-                InternalAppendOrUpdateMessage(systemMessageDto);
+                InternalAppendOrUpdateMessage(systemMessageDto, out _);
             }
 
             Truncated?.Invoke(this);
@@ -1096,10 +1105,7 @@ private Task InternalBanUserAsync(IStreamUser user, bool isShadowBan = false, st
             });
         }
 
-        private void SortMessagesByCreatedAt()
-        {
-            _messages.Sort((msg1, msg2) => msg1.CreatedAt.CompareTo(msg2.CreatedAt));
-        }
+        private void SortMessagesByCreatedAt() => _messages.Sort(_messageCreateAtComparer);
 
         private UpdateChannelRequestInternalDTO GetUpdateRequestWithCurrentData()
             => new UpdateChannelRequestInternalDTO

Assets/Plugins/StreamChat/Core/StatefulModels/StreamChannel.cs

@@ -70,6 +70,8 @@ internal sealed class StreamMessage : StreamStatefulModelBase<StreamMessage>,
 
         public int? ReplyCount { get; private set; }
 
+        //StreamTODO: Channel filters out messages based of this field but this won't work if the message got updated
+        //Channel needs to react to message.updated event and account for already present message becoming shadowed
         public bool? Shadowed { get; private set; }
 
         public bool? ShowInChannel { get; private set; }
@@ -190,6 +192,8 @@ public Task MarkMessageAsLastReadAsync()
                 });
         }
 
+        public override string ToString() => $"{nameof(IStreamMessage)}: {Text}, From: {User}";
+
         void IUpdateableFrom<MessageInternalDTO, StreamMessage>.UpdateFromDto(MessageInternalDTO dto, ICache cache)
         {
             _attachments.TryReplaceRegularObjectsFromDto(dto.Attachments, cache);

Assets/Plugins/StreamChat/Core/StatefulModels/StreamMessage.cs

@@ -95,6 +95,8 @@ await TryCloseAndDisposeAsync(WebSocketCloseStatus.NormalClosure,
             {
                 await HandleConnectionFailedAsync(e);
                 return;
+                
+                //StreamTodo: failure should throw an exception. In our scenario this doesn't matter because the caller only logs a potential exception and handled failure based of the ConnectionFailed event only
             }
 
             _backgroundSendTimer = new Timer(SendMessagesCallback, null, 0, UpdatePeriod);

Assets/Plugins/StreamChat/Libs/Websockets/WebsocketClient.cs

@@ -0,0 +1,64 @@
+using System.Collections;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+using NUnit.Framework;
+using StreamChat.Core.LowLevelClient.Events;
+using UnityEngine.TestTools;
+
+#if STREAM_TESTS_ENABLED
+namespace StreamChat.Tests.StatefulClient
+{
+    internal class ChannelsModerationTests : BaseStateIntegrationTests
+    {
+        // This test requires active AI moderation policy - we rely on message being shadowed by the AI moderation.
+        // Set "Threat" AI filter to "shadow block"
+        // Note: The moderation won't work if you have the `Skip Message Moderation` permission set for this role. Admins probably have this by default.
+        [UnityTest]
+        public IEnumerator When_client_sends_message_shadowed_by_ai_moderation_expect_other_client_to_not_receive_it()
+            => ConnectAndExecute(When_client_sends_message_shadowed_by_ai_moderation_expect_other_client_to_not_receive_it_Async);
+
+        private async Task When_client_sends_message_shadowed_by_ai_moderation_expect_other_client_to_not_receive_it_Async()
+        {
+            // Create channel
+            var channel = await CreateUniqueTempChannelAsync();
+
+            var otherClient = await GetConnectedOtherClientAsync();
+            
+            // Fetch channel on other client to get it loaded into state layer
+            var otherClientChannel = await otherClient.GetOrCreateChannelWithIdAsync(channel.Type, channel.Id);
+
+            
+            Assert.AreEqual(channel.Cid, otherClientChannel.Cid);
+            
+            var normalMessage = await channel.SendNewMessageAsync("normal message");
+            
+            // Appropriate message should be present for both clients
+            Assert.IsTrue(channel.Messages.Contains(normalMessage));
+            
+            // Wait for other client to receive the message
+            await WaitWhileFalseAsync(() => otherClientChannel.Messages.Any(m => m.Id == normalMessage.Id));
+            
+            Assert.IsNotNull(otherClientChannel.Messages.Single(m => m.Id == normalMessage.Id));
+            
+            // Setup waiting for the message on the other client
+            var messagesReceivedOnOtherClient = new List<EventMessageNew>();
+            otherClient.InternalLowLevelClient.MessageReceived += eventMessageNew =>
+            {
+                messagesReceivedOnOtherClient.Add(eventMessageNew);
+            };
+            
+            var offensiveMessage = await channel.SendNewMessageAsync("I shall kidnap your hamster!");
+            
+            // Author should believe the message was sent
+            Assert.IsTrue(channel.Messages.Contains(offensiveMessage));
+
+            // Wait for other client to receive the offensive message
+            await WaitWhileFalseAsync(() => messagesReceivedOnOtherClient.Any(m => m.Message.Id == offensiveMessage.Id));
+            
+            // Other client should have the offensive message shadowed
+            Assert.IsFalse(otherClientChannel.Messages.Any(m => m.Id == offensiveMessage.Id));
+        }
+    }
+}
+#endif