Merge pull request #124 from NicolasConstant/develop

0.19.0 PR

VARIABLES.md

@@ -48,6 +48,7 @@ If both whitelisting and blacklisting are set, only the whitelisting will be act
 * `Instance:PublishReplies` (default: false) to enable or disable replies publishing.
 * `Instance:UnlistedTwitterAccounts` (default: null) to enable unlisted publication for selected twitter accounts, separated by `;` (please limit this to brands and other public profiles). 
 * `Instance:SensitiveTwitterAccounts` (default: null) mark all media from given accounts as sensitive by default, separated by `;`. 
+* `Instance:FailingTwitterUserCleanUpThreshold` (default: 700) set the max allowed errors (due to a banned/deleted/private account) from a Twitter Account retrieval before auto-removal. (by default an account is called every 15 mins)
 
 # Docker Compose full example
 

src/BSLManager/App.cs

@@ -146,23 +146,31 @@ namespace BSLManager
                 Width = Dim.Fill(),
                 Height = 1
             };
-            var inbox = new Label($"Inbox: {follower.InboxRoute}")
+            var errors = new Label($"Posting Errors: {follower.PostingErrorCount}")
             {
                 X = 1,
                 Y = 4,
                 Width = Dim.Fill(),
                 Height = 1
             };
-            var sharedInbox = new Label($"Shared Inbox: {follower.SharedInboxRoute}")
+            var inbox = new Label($"Inbox: {follower.InboxRoute}")
             {
                 X = 1,
                 Y = 5,
                 Width = Dim.Fill(),
                 Height = 1
             };
+            var sharedInbox = new Label($"Shared Inbox: {follower.SharedInboxRoute}")
+            {
+                X = 1,
+                Y = 6,
+                Width = Dim.Fill(),
+                Height = 1
+            };
 
             dialog.Add(name);
             dialog.Add(following);
+            dialog.Add(errors);
             dialog.Add(inbox);
             dialog.Add(sharedInbox);
             dialog.Add(close);

src/BSLManager/Domain/FollowersListState.cs

@@ -26,7 +26,7 @@ namespace BSLManager.Domain
 
             foreach (var follower in _sourceUserList)
             {
-                var displayedUser = $"{GetFullHandle(follower)}     ({follower.Followings.Count})";
+                var displayedUser = $"{GetFullHandle(follower)}   ({follower.Followings.Count})  (err:{follower.PostingErrorCount})";
                 _filteredDisplayableUserList.Add(displayedUser);
             }
         }

src/BirdsiteLive.Common/Settings/InstanceSettings.cs

@@ -11,5 +11,7 @@
 
         public string UnlistedTwitterAccounts { get; set; }
         public string SensitiveTwitterAccounts { get; set; }
+
+        public int FailingTwitterUserCleanUpThreshold { get; set; }
     }
 }

src/BirdsiteLive.Pipeline/BirdsiteLive.Pipeline.csproj

@@ -13,6 +13,7 @@
 
   <ItemGroup>
     <ProjectReference Include="..\BirdsiteLive.Domain\BirdsiteLive.Domain.csproj" />
+    <ProjectReference Include="..\BirdsiteLive.Moderation\BirdsiteLive.Moderation.csproj" />
     <ProjectReference Include="..\BirdsiteLive.Twitter\BirdsiteLive.Twitter.csproj" />
     <ProjectReference Include="..\DataAccessLayers\BirdsiteLive.DAL\BirdsiteLive.DAL.csproj" />
   </ItemGroup>

src/BirdsiteLive.Pipeline/Contracts/IRefreshTwitterUserStatusProcessor.cs

@@ -0,0 +1,12 @@
+using System.Threading;
+using System.Threading.Tasks;
+using BirdsiteLive.DAL.Models;
+using BirdsiteLive.Pipeline.Models;
+
+namespace BirdsiteLive.Pipeline.Contracts
+{
+    public interface IRefreshTwitterUserStatusProcessor
+    {
+        Task<UserWithDataToSync[]> ProcessAsync(SyncTwitterUser[] syncTwitterUsers, CancellationToken ct);
+    }
+}

src/BirdsiteLive.Pipeline/Contracts/IRetrieveFollowersProcessor.cs

@@ -7,7 +7,7 @@ namespace BirdsiteLive.Pipeline.Contracts
 {
     public interface IRetrieveFollowersProcessor
     {
-        Task<IEnumerable<UserWithTweetsToSync>> ProcessAsync(UserWithTweetsToSync[] userWithTweetsToSyncs, CancellationToken ct);
+        Task<IEnumerable<UserWithDataToSync>> ProcessAsync(UserWithDataToSync[] userWithTweetsToSyncs, CancellationToken ct);
         //IAsyncEnumerable<UserWithTweetsToSync> ProcessAsync(UserWithTweetsToSync[] userWithTweetsToSyncs, CancellationToken ct);
     }
 }

src/BirdsiteLive.Pipeline/Contracts/IRetrieveTweetsProcessor.cs

@@ -7,6 +7,6 @@ namespace BirdsiteLive.Pipeline.Contracts
 {
     public interface IRetrieveTweetsProcessor
     {
-        Task<UserWithTweetsToSync[]> ProcessAsync(SyncTwitterUser[] syncTwitterUsers, CancellationToken ct);
+        Task<UserWithDataToSync[]> ProcessAsync(UserWithDataToSync[] syncTwitterUsers, CancellationToken ct);
     }
 }

src/BirdsiteLive.Pipeline/Contracts/ISaveProgressionProcessor.cs

@@ -6,6 +6,6 @@ namespace BirdsiteLive.Pipeline.Contracts
 {
     public interface ISaveProgressionProcessor
     {
-        Task ProcessAsync(UserWithTweetsToSync userWithTweetsToSync, CancellationToken ct);
+        Task ProcessAsync(UserWithDataToSync userWithTweetsToSync, CancellationToken ct);
     }
 }

src/BirdsiteLive.Pipeline/Contracts/ISendTweetsToFollowersProcessor.cs

@@ -6,6 +6,6 @@ namespace BirdsiteLive.Pipeline.Contracts
 {
     public interface ISendTweetsToFollowersProcessor
     {
-        Task<UserWithTweetsToSync> ProcessAsync(UserWithTweetsToSync userWithTweetsToSync, CancellationToken ct);
+        Task<UserWithDataToSync> ProcessAsync(UserWithDataToSync userWithTweetsToSync, CancellationToken ct);
     }
 }

src/BirdsiteLive.Pipeline/Models/UserWithDataToSync.cs

@@ -4,7 +4,7 @@ using Tweetinvi.Models;
 
 namespace BirdsiteLive.Pipeline.Models
 {
-    public class UserWithTweetsToSync
+    public class UserWithDataToSync
     {
         public SyncTwitterUser User { get; set; }
         public ExtractedTweet[] Tweets { get; set; }

src/BirdsiteLive.Pipeline/Processors/RefreshTwitterUserStatusProcessor.cs

@@ -0,0 +1,74 @@
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using BirdsiteLive.Common.Settings;
+using BirdsiteLive.DAL.Contracts;
+using BirdsiteLive.DAL.Models;
+using BirdsiteLive.Moderation.Actions;
+using BirdsiteLive.Pipeline.Contracts;
+using BirdsiteLive.Pipeline.Models;
+using BirdsiteLive.Twitter;
+
+namespace BirdsiteLive.Pipeline.Processors
+{
+    public class RefreshTwitterUserStatusProcessor : IRefreshTwitterUserStatusProcessor
+    {
+        private readonly ICachedTwitterUserService _twitterUserService;
+        private readonly ITwitterUserDal _twitterUserDal;
+        private readonly IRemoveTwitterAccountAction _removeTwitterAccountAction;
+        private readonly InstanceSettings _instanceSettings;
+
+        #region Ctor
+        public RefreshTwitterUserStatusProcessor(ICachedTwitterUserService twitterUserService, ITwitterUserDal twitterUserDal, IRemoveTwitterAccountAction removeTwitterAccountAction, InstanceSettings instanceSettings)
+        {
+            _twitterUserService = twitterUserService;
+            _twitterUserDal = twitterUserDal;
+            _removeTwitterAccountAction = removeTwitterAccountAction;
+            _instanceSettings = instanceSettings;
+        }
+        #endregion
+
+        public async Task<UserWithDataToSync[]> ProcessAsync(SyncTwitterUser[] syncTwitterUsers, CancellationToken ct)
+        {
+            var usersWtData = new List<UserWithDataToSync>();
+
+            foreach (var user in syncTwitterUsers)
+            {
+                var userView = _twitterUserService.GetUser(user.Acct);
+                if (userView == null)
+                {
+                    await AnalyseFailingUserAsync(user);
+                }
+                else if (!userView.Protected)
+                {
+                    user.FetchingErrorCount = 0;
+                    var userWtData = new UserWithDataToSync
+                    {
+                        User = user
+                    };
+                    usersWtData.Add(userWtData);
+                }
+            }
+
+            return usersWtData.ToArray();
+        }
+
+        private async Task AnalyseFailingUserAsync(SyncTwitterUser user)
+        {
+            var dbUser = await _twitterUserDal.GetTwitterUserAsync(user.Acct);
+            dbUser.FetchingErrorCount++;
+
+            if (dbUser.FetchingErrorCount > _instanceSettings.FailingTwitterUserCleanUpThreshold)
+            {
+                await _removeTwitterAccountAction.ProcessAsync(user);
+            }
+            else
+            {
+                await _twitterUserDal.UpdateTwitterUserAsync(dbUser);
+            }
+
+            // Purge
+            _twitterUserService.PurgeUser(user.Acct);
+        }
+    }
+}

src/BirdsiteLive.Pipeline/Processors/RetrieveFollowersProcessor.cs

@@ -18,7 +18,7 @@ namespace BirdsiteLive.Pipeline.Processors
         }
         #endregion
 
-        public async Task<IEnumerable<UserWithTweetsToSync>> ProcessAsync(UserWithTweetsToSync[] userWithTweetsToSyncs, CancellationToken ct)
+        public async Task<IEnumerable<UserWithDataToSync>> ProcessAsync(UserWithDataToSync[] userWithTweetsToSyncs, CancellationToken ct)
         {
             //TODO multithread this
             foreach (var user in userWithTweetsToSyncs)

src/BirdsiteLive.Pipeline/Processors/RetrieveTweetsProcessor.cs

@@ -31,33 +31,30 @@ namespace BirdsiteLive.Pipeline.Processors
         }
         #endregion
 
-        public async Task<UserWithTweetsToSync[]> ProcessAsync(SyncTwitterUser[] syncTwitterUsers, CancellationToken ct)
+        public async Task<UserWithDataToSync[]> ProcessAsync(UserWithDataToSync[] syncTwitterUsers, CancellationToken ct)
         {
-            var usersWtTweets = new List<UserWithTweetsToSync>();
+            var usersWtTweets = new List<UserWithDataToSync>();
 
             //TODO multithread this
-            foreach (var user in syncTwitterUsers)
+            foreach (var userWtData in syncTwitterUsers)
             {
+                var user = userWtData.User;
                 var tweets = RetrieveNewTweets(user);
                 if (tweets.Length > 0 && user.LastTweetPostedId != -1)
                 {
-                    var userWtTweets = new UserWithTweetsToSync
-                    {
-                        User = user,
-                        Tweets = tweets
-                    };
-                    usersWtTweets.Add(userWtTweets);
+                    userWtData.Tweets = tweets;
+                    usersWtTweets.Add(userWtData);
                 }
                 else if (tweets.Length > 0 && user.LastTweetPostedId == -1)
                 {
                     var tweetId = tweets.Last().Id;
                     var now = DateTime.UtcNow;
-                    await _twitterUserDal.UpdateTwitterUserAsync(user.Id, tweetId, tweetId, now);
+                    await _twitterUserDal.UpdateTwitterUserAsync(user.Id, tweetId, tweetId, user.FetchingErrorCount, now);
                 }
                 else
                 {
                     var now = DateTime.UtcNow;
-                    await _twitterUserDal.UpdateTwitterUserAsync(user.Id, user.LastTweetPostedId, user.LastTweetSynchronizedForAllFollowersId, now);
+                    await _twitterUserDal.UpdateTwitterUserAsync(user.Id, user.LastTweetPostedId, user.LastTweetSynchronizedForAllFollowersId, user.FetchingErrorCount, now);
                 }
             }
 
@@ -67,11 +64,7 @@ namespace BirdsiteLive.Pipeline.Processors
         private ExtractedTweet[] RetrieveNewTweets(SyncTwitterUser user)
         {
             var tweets = new ExtractedTweet[0];
-
-            // Don't retrieve TL if protected
-            var userView = _twitterUserService.GetUser(user.Acct);
-            if (userView == null || userView.Protected) return tweets;
-
+            
             try
             {
                 if (user.LastTweetPostedId == -1)

src/BirdsiteLive.Pipeline/Processors/RetrieveTwitterUsersProcessor.cs

@@ -55,7 +55,12 @@ namespace BirdsiteLive.Pipeline.Processors
                     }
 
                     var splitCount = splitUsers.Count();
-                    if (splitCount < 15) await Task.Delay((15 - splitCount) * WaitFactor, ct);
+                    if (splitCount < 15) await Task.Delay((15 - splitCount) * WaitFactor, ct); //Always wait 15min
+
+                    //// Extra wait time to fit 100.000/day limit
+                    //var extraWaitTime = (int)Math.Ceiling((60 / ((100000d / 24) / userCount)) - 15);
+                    //if (extraWaitTime < 0) extraWaitTime = 0;
+                    //await Task.Delay(extraWaitTime * 1000, ct);
                 }
                 catch (Exception e)
                 {

src/BirdsiteLive.Pipeline/Processors/SaveProgressionProcessor.cs

@@ -22,7 +22,7 @@ namespace BirdsiteLive.Pipeline.Processors
         }
         #endregion
 
-        public async Task ProcessAsync(UserWithTweetsToSync userWithTweetsToSync, CancellationToken ct)
+        public async Task ProcessAsync(UserWithDataToSync userWithTweetsToSync, CancellationToken ct)
         {
             try
             {
@@ -49,7 +49,7 @@ namespace BirdsiteLive.Pipeline.Processors
                 var lastPostedTweet = userWithTweetsToSync.Tweets.Select(x => x.Id).Max();
                 var minimumSync = followingSyncStatuses.Min();
                 var now = DateTime.UtcNow;
-                await _twitterUserDal.UpdateTwitterUserAsync(userId, lastPostedTweet, minimumSync, now);
+                await _twitterUserDal.UpdateTwitterUserAsync(userId, lastPostedTweet, minimumSync, userWithTweetsToSync.User.FetchingErrorCount, now);
             }
             catch (Exception e)
             {

src/BirdsiteLive.Pipeline/Processors/SendTweetsToFollowersProcessor.cs

@@ -22,18 +22,20 @@ namespace BirdsiteLive.Pipeline.Processors
     {
         private readonly ISendTweetsToInboxTask _sendTweetsToInboxTask;
         private readonly ISendTweetsToSharedInboxTask _sendTweetsToSharedInbox;
+        private readonly IFollowersDal _followersDal;
         private readonly ILogger<SendTweetsToFollowersProcessor> _logger;
 
         #region Ctor
-        public SendTweetsToFollowersProcessor(ISendTweetsToInboxTask sendTweetsToInboxTask, ISendTweetsToSharedInboxTask sendTweetsToSharedInbox, ILogger<SendTweetsToFollowersProcessor> logger)
+        public SendTweetsToFollowersProcessor(ISendTweetsToInboxTask sendTweetsToInboxTask, ISendTweetsToSharedInboxTask sendTweetsToSharedInbox, IFollowersDal followersDal, ILogger<SendTweetsToFollowersProcessor> logger)
         {
             _sendTweetsToInboxTask = sendTweetsToInboxTask;
             _sendTweetsToSharedInbox = sendTweetsToSharedInbox;
             _logger = logger;
+            _followersDal = followersDal;
         }
         #endregion
 
-        public async Task<UserWithTweetsToSync> ProcessAsync(UserWithTweetsToSync userWithTweetsToSync, CancellationToken ct)
+        public async Task<UserWithDataToSync> ProcessAsync(UserWithDataToSync userWithTweetsToSync, CancellationToken ct)
         {
             var user = userWithTweetsToSync.User;
 
@@ -41,18 +43,18 @@ namespace BirdsiteLive.Pipeline.Processors
             var followersWtSharedInbox = userWithTweetsToSync.Followers
                 .Where(x => !string.IsNullOrWhiteSpace(x.SharedInboxRoute))
                 .ToList();
-            await ProcessFollowersWithSharedInbox(userWithTweetsToSync.Tweets, followersWtSharedInbox, user);
+            await ProcessFollowersWithSharedInboxAsync(userWithTweetsToSync.Tweets, followersWtSharedInbox, user);
 
             // Process Inbox
             var followerWtInbox = userWithTweetsToSync.Followers
                 .Where(x => string.IsNullOrWhiteSpace(x.SharedInboxRoute))
                 .ToList();
-            await ProcessFollowersWithInbox(userWithTweetsToSync.Tweets, followerWtInbox, user);
+            await ProcessFollowersWithInboxAsync(userWithTweetsToSync.Tweets, followerWtInbox, user);
 
             return userWithTweetsToSync;
         }
 
-        private async Task ProcessFollowersWithSharedInbox(ExtractedTweet[] tweets, List<Follower> followers, SyncTwitterUser user)
+        private async Task ProcessFollowersWithSharedInboxAsync(ExtractedTweet[] tweets, List<Follower> followers, SyncTwitterUser user)
         {
             var followersPerInstances = followers.GroupBy(x => x.Host);
 
@@ -61,28 +63,51 @@ namespace BirdsiteLive.Pipeline.Processors
                 try
                 {
                     await _sendTweetsToSharedInbox.ExecuteAsync(tweets, user, followersPerInstance.Key, followersPerInstance.ToArray());
+
+                    foreach (var f in followersPerInstance)
+                        await ProcessWorkingUserAsync(f);
                 }
                 catch (Exception e)
                 {
                     var follower = followersPerInstance.First();
                     _logger.LogError(e, "Posting to {Host}{Route} failed", follower.Host, follower.SharedInboxRoute);
+
+                    foreach (var f in followersPerInstance)
+                        await ProcessFailingUserAsync(f);
                 }
             }
         }
         
-        private async Task ProcessFollowersWithInbox(ExtractedTweet[] tweets, List<Follower> followerWtInbox, SyncTwitterUser user)
+        private async Task ProcessFollowersWithInboxAsync(ExtractedTweet[] tweets, List<Follower> followerWtInbox, SyncTwitterUser user)
         {
             foreach (var follower in followerWtInbox)
             {
                 try
                 {
                     await _sendTweetsToInboxTask.ExecuteAsync(tweets, follower, user);
+                    await ProcessWorkingUserAsync(follower);
                 }
                 catch (Exception e)
                 {
                     _logger.LogError(e, "Posting to {Host}{Route} failed", follower.Host, follower.InboxRoute);
+                    await ProcessFailingUserAsync(follower);
                 }
             }
         }
+
+        private async Task ProcessWorkingUserAsync(Follower follower)
+        {
+            if (follower.PostingErrorCount > 0)
+            {
+                follower.PostingErrorCount = 0;
+                await _followersDal.UpdateFollowerAsync(follower);
+            }
+        }
+
+        private async Task ProcessFailingUserAsync(Follower follower)
+        {
+            follower.PostingErrorCount++;
+            await _followersDal.UpdateFollowerAsync(follower);
+        }
     }
 }

src/BirdsiteLive.Pipeline/StatusPublicationPipeline.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Runtime.CompilerServices;
 using System.Threading;
 using System.Threading.Tasks;
 using System.Threading.Tasks.Dataflow;
@@ -17,6 +18,7 @@ namespace BirdsiteLive.Pipeline
     public class StatusPublicationPipeline : IStatusPublicationPipeline
     {
         private readonly IRetrieveTwitterUsersProcessor _retrieveTwitterAccountsProcessor;
+        private readonly IRefreshTwitterUserStatusProcessor _refreshTwitterUserStatusProcessor;
         private readonly IRetrieveTweetsProcessor _retrieveTweetsProcessor;
         private readonly IRetrieveFollowersProcessor _retrieveFollowersProcessor;
         private readonly ISendTweetsToFollowersProcessor _sendTweetsToFollowersProcessor;
@@ -24,13 +26,14 @@ namespace BirdsiteLive.Pipeline
         private readonly ILogger<StatusPublicationPipeline> _logger;
 
         #region Ctor
-        public StatusPublicationPipeline(IRetrieveTweetsProcessor retrieveTweetsProcessor, IRetrieveTwitterUsersProcessor retrieveTwitterAccountsProcessor, IRetrieveFollowersProcessor retrieveFollowersProcessor, ISendTweetsToFollowersProcessor sendTweetsToFollowersProcessor, ISaveProgressionProcessor saveProgressionProcessor, ILogger<StatusPublicationPipeline> logger)
+        public StatusPublicationPipeline(IRetrieveTweetsProcessor retrieveTweetsProcessor, IRetrieveTwitterUsersProcessor retrieveTwitterAccountsProcessor, IRetrieveFollowersProcessor retrieveFollowersProcessor, ISendTweetsToFollowersProcessor sendTweetsToFollowersProcessor, ISaveProgressionProcessor saveProgressionProcessor, IRefreshTwitterUserStatusProcessor refreshTwitterUserStatusProcessor, ILogger<StatusPublicationPipeline> logger)
         {
             _retrieveTweetsProcessor = retrieveTweetsProcessor;
             _retrieveTwitterAccountsProcessor = retrieveTwitterAccountsProcessor;
             _retrieveFollowersProcessor = retrieveFollowersProcessor;
             _sendTweetsToFollowersProcessor = sendTweetsToFollowersProcessor;
             _saveProgressionProcessor = saveProgressionProcessor;
+            _refreshTwitterUserStatusProcessor = refreshTwitterUserStatusProcessor;
 
             _logger = logger;
         }
@@ -39,16 +42,21 @@ namespace BirdsiteLive.Pipeline
         public async Task ExecuteAsync(CancellationToken ct)
         {
             // Create blocks 
-            var twitterUsersBufferBlock = new BufferBlock<SyncTwitterUser[]>(new DataflowBlockOptions { BoundedCapacity = 1, CancellationToken = ct });
-            var retrieveTweetsBlock = new TransformBlock<SyncTwitterUser[], UserWithTweetsToSync[]>(async x => await _retrieveTweetsProcessor.ProcessAsync(x, ct));
-            var retrieveTweetsBufferBlock = new BufferBlock<UserWithTweetsToSync[]>(new DataflowBlockOptions { BoundedCapacity = 1, CancellationToken = ct });
-            var retrieveFollowersBlock = new TransformManyBlock<UserWithTweetsToSync[], UserWithTweetsToSync>(async x => await _retrieveFollowersProcessor.ProcessAsync(x, ct));
-            var retrieveFollowersBufferBlock = new BufferBlock<UserWithTweetsToSync>(new DataflowBlockOptions { BoundedCapacity = 20, CancellationToken = ct });
-            var sendTweetsToFollowersBlock = new TransformBlock<UserWithTweetsToSync, UserWithTweetsToSync>(async x => await _sendTweetsToFollowersProcessor.ProcessAsync(x, ct), new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 5, CancellationToken = ct });
-            var sendTweetsToFollowersBufferBlock = new BufferBlock<UserWithTweetsToSync>(new DataflowBlockOptions { BoundedCapacity = 20, CancellationToken = ct });
-            var saveProgressionBlock = new ActionBlock<UserWithTweetsToSync>(async x => await _saveProgressionProcessor.ProcessAsync(x, ct), new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 5, CancellationToken = ct });
+            var twitterUserToRefreshBufferBlock = new BufferBlock<SyncTwitterUser[]>(new DataflowBlockOptions
+                { BoundedCapacity = 1, CancellationToken = ct });
+            var twitterUserToRefreshBlock = new TransformBlock<SyncTwitterUser[], UserWithDataToSync[]>(async x => await _refreshTwitterUserStatusProcessor.ProcessAsync(x, ct));
+            var twitterUsersBufferBlock = new BufferBlock<UserWithDataToSync[]>(new DataflowBlockOptions { BoundedCapacity = 1, CancellationToken = ct });
+            var retrieveTweetsBlock = new TransformBlock<UserWithDataToSync[], UserWithDataToSync[]>(async x => await _retrieveTweetsProcessor.ProcessAsync(x, ct));
+            var retrieveTweetsBufferBlock = new BufferBlock<UserWithDataToSync[]>(new DataflowBlockOptions { BoundedCapacity = 1, CancellationToken = ct });
+            var retrieveFollowersBlock = new TransformManyBlock<UserWithDataToSync[], UserWithDataToSync>(async x => await _retrieveFollowersProcessor.ProcessAsync(x, ct));
+            var retrieveFollowersBufferBlock = new BufferBlock<UserWithDataToSync>(new DataflowBlockOptions { BoundedCapacity = 20, CancellationToken = ct });
+            var sendTweetsToFollowersBlock = new TransformBlock<UserWithDataToSync, UserWithDataToSync>(async x => await _sendTweetsToFollowersProcessor.ProcessAsync(x, ct), new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 5, CancellationToken = ct });
+            var sendTweetsToFollowersBufferBlock = new BufferBlock<UserWithDataToSync>(new DataflowBlockOptions { BoundedCapacity = 20, CancellationToken = ct });
+            var saveProgressionBlock = new ActionBlock<UserWithDataToSync>(async x => await _saveProgressionProcessor.ProcessAsync(x, ct), new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 5, CancellationToken = ct });
 
             // Link pipeline
+            twitterUserToRefreshBufferBlock.LinkTo(twitterUserToRefreshBlock, new DataflowLinkOptions { PropagateCompletion = true });
+            twitterUserToRefreshBlock.LinkTo(twitterUsersBufferBlock, new DataflowLinkOptions { PropagateCompletion = true });
             twitterUsersBufferBlock.LinkTo(retrieveTweetsBlock, new DataflowLinkOptions { PropagateCompletion = true });
             retrieveTweetsBlock.LinkTo(retrieveTweetsBufferBlock, new DataflowLinkOptions { PropagateCompletion = true });
             retrieveTweetsBufferBlock.LinkTo(retrieveFollowersBlock, new DataflowLinkOptions { PropagateCompletion = true });
@@ -58,7 +66,7 @@ namespace BirdsiteLive.Pipeline
             sendTweetsToFollowersBufferBlock.LinkTo(saveProgressionBlock, new DataflowLinkOptions { PropagateCompletion = true });
 
             // Launch twitter user retriever
-            var retrieveTwitterAccountsTask = _retrieveTwitterAccountsProcessor.GetTwitterUsersAsync(twitterUsersBufferBlock, ct);
+            var retrieveTwitterAccountsTask = _retrieveTwitterAccountsProcessor.GetTwitterUsersAsync(twitterUserToRefreshBufferBlock, ct);
 
             // Wait
             await Task.WhenAny(new[] { retrieveTwitterAccountsTask, saveProgressionBlock.Completion });

src/BirdsiteLive.Twitter/Statistics/TwitterStatisticsHandler.cs

@@ -95,7 +95,7 @@ namespace BirdsiteLive.Statistics.Domain
                 TimelineCallsCountMin = timelineCalls.Any() ? timelineCalls.Min() : 0,
                 TimelineCallsCountAvg = timelineCalls.Any() ? (int)timelineCalls.Average() : 0,
                 TimelineCallsCountMax = timelineCalls.Any() ? timelineCalls.Max() : 0,
-                TimelineCallsMax = 1500
+                TimelineCallsMax = 1000
             };
         }
     }

src/BirdsiteLive.Twitter/TwitterUserService.cs

@@ -49,6 +49,9 @@ namespace BirdsiteLive.Twitter
             catch (Exception e)
             {
                 _logger.LogError(e, "Error retrieving user {Username}", username);
+
+                // TODO keep track of error, see where to remove user if too much errors
+
                 return null;
             }
 

src/BirdsiteLive/BirdsiteLive.csproj

@@ -4,7 +4,7 @@
     <TargetFramework>netcoreapp3.1</TargetFramework>
     <UserSecretsId>d21486de-a812-47eb-a419-05682bb68856</UserSecretsId>
     <DockerDefaultTargetOS>Linux</DockerDefaultTargetOS>
-    <Version>0.18.0</Version>
+    <Version>0.19.0</Version>
   </PropertyGroup>
 
   <ItemGroup>

src/BirdsiteLive/Controllers/StatisticsController.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
+using System.Threading;
 using System.Threading.Tasks;
 using BirdsiteLive.DAL.Contracts;
 using BirdsiteLive.Domain.Statistics;
@@ -16,7 +17,6 @@ namespace BirdsiteLive.Controllers
         private readonly ITwitterStatisticsHandler _twitterStatistics;
         private readonly IExtractionStatisticsHandler _extractionStatistics;
 
-
         #region Ctor
         public StatisticsController(ITwitterUserDal twitterUserDal, IFollowersDal followersDal, ITwitterStatisticsHandler twitterStatistics, IExtractionStatisticsHandler extractionStatistics)
         {
@@ -32,7 +32,9 @@ namespace BirdsiteLive.Controllers
             var stats = new Models.StatisticsModels.Statistics
             {
                 FollowersCount = await _followersDal.GetFollowersCountAsync(),
+                FailingFollowersCount = await _followersDal.GetFailingFollowersCountAsync(),
                 TwitterUserCount = await _twitterUserDal.GetTwitterUsersCountAsync(),
+                FailingTwitterUserCount = await _twitterUserDal.GetFailingTwitterUsersCountAsync(),
                 TwitterStatistics = _twitterStatistics.GetStatistics(),
                 ExtractionStatistics = _extractionStatistics.GetStatistics(),
             };

src/BirdsiteLive/Controllers/UsersController.cs

@@ -58,6 +58,7 @@ namespace BirdsiteLive.Controllers
 
         [Route("/@{id}")]
         [Route("/users/{id}")]
+        [Route("/users/{id}/remote_follow")]
         public IActionResult Index(string id)
         {
             _logger.LogTrace("User Index: {Id}", id);

src/BirdsiteLive/Models/StatisticsModels/Statistics.cs

@@ -6,7 +6,9 @@ namespace BirdsiteLive.Models.StatisticsModels
     public class Statistics
     {
         public int FollowersCount { get; set; }
+        public int FailingFollowersCount { get; set; }
         public int TwitterUserCount { get; set; }
+        public int FailingTwitterUserCount { get; set; }
         public ApiStatistics TwitterStatistics { get; set; }
         public ExtractionStatistics ExtractionStatistics { get; set; }
     }

src/BirdsiteLive/Views/Statistics/Index.cshtml

@@ -9,7 +9,9 @@
 <h4>Instance</h4>
 <ul>
     <li>Twitter Users: @Model.TwitterUserCount</li>
+    <li>Failing Twitter Users: @Model.FailingTwitterUserCount</li>
     <li>Followers: @Model.FollowersCount</li>
+    <li>Failing Followers: @Model.FailingFollowersCount</li>
 </ul>
 
 <h4>Twitter API (Min, Avg, Max for the last 24h)</h4>

src/BirdsiteLive/appsettings.json

@@ -20,9 +20,10 @@
     "AdminEmail": "me@domain.name",
     "ResolveMentionsInProfiles": true,
     "PublishReplies": false,
-    "MaxUsersCapacity": 1500,
+    "MaxUsersCapacity": 1000,
     "UnlistedTwitterAccounts": null,
-    "SensitiveTwitterAccounts": null
+    "SensitiveTwitterAccounts": null,
+    "FailingTwitterUserCleanUpThreshold": 700
   },
   "Db": {
     "Type": "postgres",

src/DataAccessLayers/BirdsiteLive.DAL.Postgres/DataAccessLayers/DbInitializerPostgresDal.cs

@@ -23,7 +23,7 @@ namespace BirdsiteLive.DAL.Postgres.DataAccessLayers
     public class DbInitializerPostgresDal : PostgresBase, IDbInitializerDal
     {
         private readonly PostgresTools _tools;
-        private readonly Version _currentVersion = new Version(2, 1);
+        private readonly Version _currentVersion = new Version(2, 3);
         private const string DbVersionType = "db-version";
 
         #region Ctor
@@ -132,7 +132,9 @@ namespace BirdsiteLive.DAL.Postgres.DataAccessLayers
             return new[]
             {
                 new Tuple<Version, Version>(new Version(1,0), new Version(2,0)),
-                new Tuple<Version, Version>(new Version(2,0), new Version(2,1))
+                new Tuple<Version, Version>(new Version(2,0), new Version(2,1)),
+                new Tuple<Version, Version>(new Version(2,1), new Version(2,2)),
+                new Tuple<Version, Version>(new Version(2,2), new Version(2,3))
             };
         }
 
@@ -151,6 +153,16 @@ namespace BirdsiteLive.DAL.Postgres.DataAccessLayers
                 var addActorId = $@"ALTER TABLE {_settings.FollowersTableName} ADD actorId VARCHAR(2048)";
                 await _tools.ExecuteRequestAsync(addActorId);
             }
+            else if (from == new Version(2, 1) && to == new Version(2, 2))
+            {
+                var addLastSync = $@"ALTER TABLE {_settings.TwitterUserTableName} ADD fetchingErrorCount SMALLINT";
+                await _tools.ExecuteRequestAsync(addLastSync);
+            }
+            else if (from == new Version(2, 2) && to == new Version(2, 3))
+            {
+                var addPostingError = $@"ALTER TABLE {_settings.FollowersTableName} ADD postingErrorCount SMALLINT";
+                await _tools.ExecuteRequestAsync(addPostingError);
+            }
             else
             {
                 throw new NotImplementedException();

src/DataAccessLayers/BirdsiteLive.DAL.Postgres/DataAccessLayers/FollowersPostgresDal.cs

@@ -53,6 +53,19 @@ namespace BirdsiteLive.DAL.Postgres.DataAccessLayers
             }
         }
 
+        public async Task<int> GetFailingFollowersCountAsync()
+        {
+            var query = $"SELECT COUNT(*) FROM {_settings.FollowersTableName} WHERE postingErrorCount > 0";
+
+            using (var dbConnection = Connection)
+            {
+                dbConnection.Open();
+
+                var result = (await dbConnection.QueryAsync<int>(query)).FirstOrDefault();
+                return result;
+            }
+        }
+
         public async Task<Follower> GetFollowerAsync(string acct, string host)
         {
             var query = $"SELECT * FROM {_settings.FollowersTableName} WHERE acct = @acct AND host = @host";
@@ -103,13 +116,13 @@ namespace BirdsiteLive.DAL.Postgres.DataAccessLayers
             if (follower.Id == default) throw new ArgumentException("id");
 
             var serializedDic = JsonConvert.SerializeObject(follower.FollowingsSyncStatus);
-            var query = $"UPDATE {_settings.FollowersTableName} SET followings = @followings, followingsSyncStatus =  CAST(@followingsSyncStatus as json) WHERE id = @id";
+            var query = $"UPDATE {_settings.FollowersTableName} SET followings = @followings, followingsSyncStatus = CAST(@followingsSyncStatus as json), postingErrorCount = @postingErrorCount WHERE id = @id";
 
             using (var dbConnection = Connection)
             {
                 dbConnection.Open();
 
-                await dbConnection.QueryAsync(query, new { follower.Id, follower.Followings, followingsSyncStatus = serializedDic });
+                await dbConnection.QueryAsync(query, new { follower.Id, follower.Followings, followingsSyncStatus = serializedDic, postingErrorCount = follower.PostingErrorCount });
             }
         }
 
@@ -158,7 +171,8 @@ namespace BirdsiteLive.DAL.Postgres.DataAccessLayers
                 ActorId = follower.ActorId,
                 SharedInboxRoute = follower.SharedInboxRoute,
                 Followings = follower.Followings.ToList(),
-                FollowingsSyncStatus = JsonConvert.DeserializeObject<Dictionary<int,long>>(follower.FollowingsSyncStatus)
+                FollowingsSyncStatus = JsonConvert.DeserializeObject<Dictionary<int,long>>(follower.FollowingsSyncStatus),
+                PostingErrorCount = follower.PostingErrorCount
             };
         }
     }
@@ -174,5 +188,6 @@ namespace BirdsiteLive.DAL.Postgres.DataAccessLayers
         public string InboxRoute { get; set; }
         public string SharedInboxRoute { get; set; }
         public string ActorId { get; set; }
+        public int PostingErrorCount { get; set; }
     }
 }

src/DataAccessLayers/BirdsiteLive.DAL.Postgres/DataAccessLayers/TwitterUserPostgresDal.cs

@@ -73,6 +73,19 @@ namespace BirdsiteLive.DAL.Postgres.DataAccessLayers
             }
         }
 
+        public async Task<int> GetFailingTwitterUsersCountAsync()
+        {
+            var query = $"SELECT COUNT(*) FROM {_settings.TwitterUserTableName} WHERE fetchingErrorCount > 0";
+
+            using (var dbConnection = Connection)
+            {
+                dbConnection.Open();
+
+                var result = (await dbConnection.QueryAsync<int>(query)).FirstOrDefault();
+                return result;
+            }
+        }
+
         public async Task<SyncTwitterUser[]> GetAllTwitterUsersAsync(int maxNumber)
         {
             var query = $"SELECT * FROM {_settings.TwitterUserTableName} ORDER BY lastSync ASC LIMIT @maxNumber";
@@ -99,23 +112,28 @@ namespace BirdsiteLive.DAL.Postgres.DataAccessLayers
             }
         }
 
-        public async Task UpdateTwitterUserAsync(int id, long lastTweetPostedId, long lastTweetSynchronizedForAllFollowersId, DateTime lastSync)
+        public async Task UpdateTwitterUserAsync(int id, long lastTweetPostedId, long lastTweetSynchronizedForAllFollowersId, int fetchingErrorCount, DateTime lastSync)
         {
             if(id == default) throw new ArgumentException("id");
             if(lastTweetPostedId == default) throw new ArgumentException("lastTweetPostedId");
             if(lastTweetSynchronizedForAllFollowersId == default) throw new ArgumentException("lastTweetSynchronizedForAllFollowersId");
             if(lastSync == default) throw new ArgumentException("lastSync");
 
-            var query = $"UPDATE {_settings.TwitterUserTableName} SET lastTweetPostedId = @lastTweetPostedId, lastTweetSynchronizedForAllFollowersId = @lastTweetSynchronizedForAllFollowersId, lastSync = @lastSync WHERE id = @id";
+            var query = $"UPDATE {_settings.TwitterUserTableName} SET lastTweetPostedId = @lastTweetPostedId, lastTweetSynchronizedForAllFollowersId = @lastTweetSynchronizedForAllFollowersId, fetchingErrorCount = @fetchingErrorCount, lastSync = @lastSync WHERE id = @id";
 
             using (var dbConnection = Connection)
             {
                 dbConnection.Open();
 
-                await dbConnection.QueryAsync(query, new { id, lastTweetPostedId,  lastTweetSynchronizedForAllFollowersId, lastSync = lastSync.ToUniversalTime() });
+                await dbConnection.QueryAsync(query, new { id, lastTweetPostedId,  lastTweetSynchronizedForAllFollowersId, fetchingErrorCount, lastSync = lastSync.ToUniversalTime() });
             }
         }
 
+        public async Task UpdateTwitterUserAsync(SyncTwitterUser user)
+        {
+            await UpdateTwitterUserAsync(user.Id, user.LastTweetPostedId, user.LastTweetSynchronizedForAllFollowersId, user.FetchingErrorCount, user.LastSync);
+        }
+
         public async Task DeleteTwitterUserAsync(string acct)
         {
             if (string.IsNullOrWhiteSpace(acct)) throw new ArgumentException("acct");

src/DataAccessLayers/BirdsiteLive.DAL/Contracts/IFollowersDal.cs

@@ -15,5 +15,6 @@ namespace BirdsiteLive.DAL.Contracts
         Task DeleteFollowerAsync(int id);
         Task DeleteFollowerAsync(string acct, string host);
         Task<int> GetFollowersCountAsync();
+        Task<int> GetFailingFollowersCountAsync();
     }
 }

src/DataAccessLayers/BirdsiteLive.DAL/Contracts/ITwitterUserDal.cs

@@ -11,9 +11,11 @@ namespace BirdsiteLive.DAL.Contracts
         Task<SyncTwitterUser> GetTwitterUserAsync(int id);
         Task<SyncTwitterUser[]> GetAllTwitterUsersAsync(int maxNumber);
         Task<SyncTwitterUser[]> GetAllTwitterUsersAsync();
-        Task UpdateTwitterUserAsync(int id, long lastTweetPostedId, long lastTweetSynchronizedForAllFollowersId, DateTime lastSync);
+        Task UpdateTwitterUserAsync(int id, long lastTweetPostedId, long lastTweetSynchronizedForAllFollowersId, int fetchingErrorCount, DateTime lastSync);
+        Task UpdateTwitterUserAsync(SyncTwitterUser user);
         Task DeleteTwitterUserAsync(string acct);
         Task DeleteTwitterUserAsync(int id);
         Task<int> GetTwitterUsersCountAsync();
+        Task<int> GetFailingTwitterUsersCountAsync();
     }
 }

src/DataAccessLayers/BirdsiteLive.DAL/Models/Follower.cs

@@ -14,5 +14,7 @@ namespace BirdsiteLive.DAL.Models
         public string Host { get; set; }
         public string InboxRoute { get; set; }
         public string SharedInboxRoute { get; set; }
+
+        public int PostingErrorCount { get; set; }
     }
 }

src/DataAccessLayers/BirdsiteLive.DAL/Models/SyncTwitterUser.cs

@@ -11,5 +11,7 @@ namespace BirdsiteLive.DAL.Models
         public long LastTweetSynchronizedForAllFollowersId { get; set; }
 
         public DateTime LastSync { get; set; }
+
+        public int FetchingErrorCount { get; set; } //TODO: update DAL
     }
 }

src/Tests/BirdsiteLive.DAL.Postgres.Tests/DataAccessLayers/FollowersPostgresDalTests.cs

@@ -54,6 +54,7 @@ namespace BirdsiteLive.DAL.Postgres.Tests.DataAccessLayers
             Assert.AreEqual(inboxRoute, result.InboxRoute);
             Assert.AreEqual(sharedInboxRoute, result.SharedInboxRoute);
             Assert.AreEqual(actorId, result.ActorId);
+            Assert.AreEqual(0, result.PostingErrorCount);
             Assert.AreEqual(following.Length, result.Followings.Count);
             Assert.AreEqual(following[0], result.Followings[0]);
             Assert.AreEqual(followingSync.Count, result.FollowingsSyncStatus.Count);
@@ -83,6 +84,7 @@ namespace BirdsiteLive.DAL.Postgres.Tests.DataAccessLayers
             Assert.AreEqual(sharedInboxRoute, result.SharedInboxRoute);
             Assert.AreEqual(0, result.Followings.Count);
             Assert.AreEqual(0, result.FollowingsSyncStatus.Count);
+            Assert.AreEqual(0, result.PostingErrorCount);
         }
 
         [TestMethod]
@@ -125,6 +127,7 @@ namespace BirdsiteLive.DAL.Postgres.Tests.DataAccessLayers
             Assert.AreEqual(followingSync.Count, result.FollowingsSyncStatus.Count);
             Assert.AreEqual(followingSync.First().Key, result.FollowingsSyncStatus.First().Key);
             Assert.AreEqual(followingSync.First().Value, result.FollowingsSyncStatus.First().Value);
+            Assert.AreEqual(0, result.PostingErrorCount);
         }
 
         [TestMethod]
@@ -234,7 +237,7 @@ namespace BirdsiteLive.DAL.Postgres.Tests.DataAccessLayers
             actorId = $"https://{host}/{acct}";
             await dal.CreateFollowerAsync(acct, host, inboxRoute, sharedInboxRoute, actorId, following, followingSync);
 
-            //User 2 
+            //User 3
             acct = "myhandle3";
             host = "domain.ext";
             following = new[] { 1 };
@@ -247,6 +250,54 @@ namespace BirdsiteLive.DAL.Postgres.Tests.DataAccessLayers
             Assert.AreEqual(3, result);
         }
 
+        [TestMethod]
+        public async Task CountFailingFollowersAsync()
+        {
+            var dal = new FollowersPostgresDal(_settings);
+
+            var result = await dal.GetFailingFollowersCountAsync();
+            Assert.AreEqual(0, result);
+
+            //User 1 
+            var acct = "myhandle1";
+            var host = "domain.ext";
+            var following = new[] { 1, 2, 3 };
+            var followingSync = new Dictionary<int, long>();
+            var inboxRoute = "/myhandle1/inbox";
+            var sharedInboxRoute = "/inbox";
+            var actorId = $"https://{host}/{acct}";
+            await dal.CreateFollowerAsync(acct, host, inboxRoute, sharedInboxRoute, actorId, following, followingSync);
+
+            //User 2 
+            acct = "myhandle2";
+            host = "domain.ext";
+            following = new[] { 2, 4, 5 };
+            inboxRoute = "/myhandle2/inbox";
+            sharedInboxRoute = "/inbox2";
+            actorId = $"https://{host}/{acct}";
+            await dal.CreateFollowerAsync(acct, host, inboxRoute, sharedInboxRoute, actorId, following, followingSync);
+
+            var follower = await dal.GetFollowerAsync(acct, host);
+            follower.PostingErrorCount = 1;
+            await dal.UpdateFollowerAsync(follower);
+
+            //User 3
+            acct = "myhandle3";
+            host = "domain.ext";
+            following = new[] { 1 };
+            inboxRoute = "/myhandle3/inbox";
+            sharedInboxRoute = "/inbox3";
+            actorId = $"https://{host}/{acct}";
+            await dal.CreateFollowerAsync(acct, host, inboxRoute, sharedInboxRoute, actorId, following, followingSync);
+
+            follower = await dal.GetFollowerAsync(acct, host);
+            follower.PostingErrorCount = 50;
+            await dal.UpdateFollowerAsync(follower);
+
+            result = await dal.GetFailingFollowersCountAsync();
+            Assert.AreEqual(2, result);
+        }
+
         [TestMethod]
         public async Task CreateUpdateAndGetFollower_Add()
         {
@@ -276,8 +327,8 @@ namespace BirdsiteLive.DAL.Postgres.Tests.DataAccessLayers
             };
             result.Followings = updatedFollowing.ToList();
             result.FollowingsSyncStatus = updatedFollowingSync;
-
-
+            result.PostingErrorCount = 10;
+            
             await dal.UpdateFollowerAsync(result);
             result = await dal.GetFollowerAsync(acct, host);
 
@@ -286,6 +337,7 @@ namespace BirdsiteLive.DAL.Postgres.Tests.DataAccessLayers
             Assert.AreEqual(updatedFollowingSync.Count, result.FollowingsSyncStatus.Count);
             Assert.AreEqual(updatedFollowingSync.First().Key, result.FollowingsSyncStatus.First().Key);
             Assert.AreEqual(updatedFollowingSync.First().Value, result.FollowingsSyncStatus.First().Value);
+            Assert.AreEqual(10, result.PostingErrorCount);
         }
 
         [TestMethod]
@@ -316,6 +368,7 @@ namespace BirdsiteLive.DAL.Postgres.Tests.DataAccessLayers
             };
             result.Followings = updatedFollowing.ToList();
             result.FollowingsSyncStatus = updatedFollowingSync;
+            result.PostingErrorCount = 5;
 
             await dal.UpdateFollowerAsync(result);
             result = await dal.GetFollowerAsync(acct, host);
@@ -325,6 +378,41 @@ namespace BirdsiteLive.DAL.Postgres.Tests.DataAccessLayers
             Assert.AreEqual(updatedFollowingSync.Count, result.FollowingsSyncStatus.Count);
             Assert.AreEqual(updatedFollowingSync.First().Key, result.FollowingsSyncStatus.First().Key);
             Assert.AreEqual(updatedFollowingSync.First().Value, result.FollowingsSyncStatus.First().Value);
+            Assert.AreEqual(5, result.PostingErrorCount);
+        }
+
+        [TestMethod]
+        public async Task CreateUpdateAndGetFollower_ResetErrorCount()
+        {
+            var acct = "myhandle";
+            var host = "domain.ext";
+            var following = new[] { 12, 19, 23 };
+            var followingSync = new Dictionary<int, long>()
+            {
+                {12, 165L},
+                {19, 166L},
+                {23, 167L}
+            };
+            var inboxRoute = "/myhandle/inbox";
+            var sharedInboxRoute = "/inbox";
+            var actorId = $"https://{host}/{acct}";
+
+            var dal = new FollowersPostgresDal(_settings);
+            await dal.CreateFollowerAsync(acct, host, inboxRoute, sharedInboxRoute, actorId, following, followingSync);
+            var result = await dal.GetFollowerAsync(acct, host);
+            Assert.AreEqual(0, result.PostingErrorCount);
+
+            result.PostingErrorCount = 5;
+
+            await dal.UpdateFollowerAsync(result);
+            result = await dal.GetFollowerAsync(acct, host);
+            Assert.AreEqual(5, result.PostingErrorCount);
+
+            result.PostingErrorCount = 0;
+
+            await dal.UpdateFollowerAsync(result);
+            result = await dal.GetFollowerAsync(acct, host);
+            Assert.AreEqual(0, result.PostingErrorCount);
         }
 
         [TestMethod]

src/Tests/BirdsiteLive.DAL.Postgres.Tests/DataAccessLayers/TwitterUserPostgresDalTests.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Diagnostics;
 using System.Threading.Tasks;
 using System.Xml;
 using BirdsiteLive.DAL.Postgres.DataAccessLayers;
@@ -47,6 +48,7 @@ namespace BirdsiteLive.DAL.Postgres.Tests.DataAccessLayers
             Assert.AreEqual(acct, result.Acct);
             Assert.AreEqual(lastTweetId, result.LastTweetPostedId);
             Assert.AreEqual(lastTweetId, result.LastTweetSynchronizedForAllFollowersId);
+            Assert.AreEqual(0, result.FetchingErrorCount);
             Assert.IsTrue(result.Id > 0);
         }
 
@@ -83,13 +85,47 @@ namespace BirdsiteLive.DAL.Postgres.Tests.DataAccessLayers
             var updatedLastTweetId = 1600L;
             var updatedLastSyncId = 1550L;
             var now = DateTime.Now;
-            await dal.UpdateTwitterUserAsync(result.Id, updatedLastTweetId, updatedLastSyncId, now);
+            var errors = 15;
+            await dal.UpdateTwitterUserAsync(result.Id, updatedLastTweetId, updatedLastSyncId, errors, now);
 
             result = await dal.GetTwitterUserAsync(acct);
 
             Assert.AreEqual(acct, result.Acct);
             Assert.AreEqual(updatedLastTweetId, result.LastTweetPostedId);
             Assert.AreEqual(updatedLastSyncId, result.LastTweetSynchronizedForAllFollowersId);
+            Assert.AreEqual(errors, result.FetchingErrorCount);
+            Assert.IsTrue(Math.Abs((now.ToUniversalTime() - result.LastSync).Milliseconds) < 100);
+        }
+
+        [TestMethod]
+        public async Task CreateUpdate2AndGetUser()
+        {
+            var acct = "myid";
+            var lastTweetId = 1548L;
+
+            var dal = new TwitterUserPostgresDal(_settings);
+
+            await dal.CreateTwitterUserAsync(acct, lastTweetId);
+            var result = await dal.GetTwitterUserAsync(acct);
+
+
+            var updatedLastTweetId = 1600L;
+            var updatedLastSyncId = 1550L;
+            var now = DateTime.Now;
+            var errors = 15;
+
+            result.LastTweetPostedId = updatedLastTweetId;
+            result.LastTweetSynchronizedForAllFollowersId = updatedLastSyncId;
+            result.FetchingErrorCount = errors;
+            result.LastSync = now;
+            await dal.UpdateTwitterUserAsync(result);
+
+            result = await dal.GetTwitterUserAsync(acct);
+
+            Assert.AreEqual(acct, result.Acct);
+            Assert.AreEqual(updatedLastTweetId, result.LastTweetPostedId);
+            Assert.AreEqual(updatedLastSyncId, result.LastTweetSynchronizedForAllFollowersId);
+            Assert.AreEqual(errors, result.FetchingErrorCount);
             Assert.IsTrue(Math.Abs((now.ToUniversalTime() - result.LastSync).Milliseconds) < 100);
         }
 
@@ -98,7 +134,7 @@ namespace BirdsiteLive.DAL.Postgres.Tests.DataAccessLayers
         public async Task Update_NoId()
         {
             var dal = new TwitterUserPostgresDal(_settings);
-            await dal.UpdateTwitterUserAsync(default, default, default, DateTime.UtcNow);
+            await dal.UpdateTwitterUserAsync(default, default, default, default, DateTime.UtcNow);
         }
 
         [TestMethod]
@@ -106,7 +142,7 @@ namespace BirdsiteLive.DAL.Postgres.Tests.DataAccessLayers
         public async Task Update_NoLastTweetPostedId()
         {
             var dal = new TwitterUserPostgresDal(_settings);
-            await dal.UpdateTwitterUserAsync(12, default, default, DateTime.UtcNow);
+            await dal.UpdateTwitterUserAsync(12, default, default, default, DateTime.UtcNow);
         }
 
         [TestMethod]
@@ -114,7 +150,7 @@ namespace BirdsiteLive.DAL.Postgres.Tests.DataAccessLayers
         public async Task Update_NoLastTweetSynchronizedForAllFollowersId()
         {
             var dal = new TwitterUserPostgresDal(_settings);
-            await dal.UpdateTwitterUserAsync(12, 9556, default, DateTime.UtcNow);
+            await dal.UpdateTwitterUserAsync(12, 9556, default, default, DateTime.UtcNow);
         }
 
         [TestMethod]
@@ -122,7 +158,7 @@ namespace BirdsiteLive.DAL.Postgres.Tests.DataAccessLayers
         public async Task Update_NoLastSync()
         {
             var dal = new TwitterUserPostgresDal(_settings);
-            await dal.UpdateTwitterUserAsync(12, 9556, 65, default);
+            await dal.UpdateTwitterUserAsync(12, 9556, 65, default, default);
         }
 
         [TestMethod]
@@ -216,7 +252,7 @@ namespace BirdsiteLive.DAL.Postgres.Tests.DataAccessLayers
             {
                 var user = allUsers[i];
                 var date = i % 2 == 0 ? oldest : newest;
-                await dal.UpdateTwitterUserAsync(user.Id, user.LastTweetPostedId, user.LastTweetSynchronizedForAllFollowersId, date);
+                await dal.UpdateTwitterUserAsync(user.Id, user.LastTweetPostedId, user.LastTweetSynchronizedForAllFollowersId, 0, date);
             }
 
             var result = await dal.GetAllTwitterUsersAsync(10);
@@ -265,5 +301,27 @@ namespace BirdsiteLive.DAL.Postgres.Tests.DataAccessLayers
             var result = await dal.GetTwitterUsersCountAsync();
             Assert.AreEqual(10, result);
         }
+
+        [TestMethod]
+        public async Task CountFailingTwitterUsers()
+        {
+            var dal = new TwitterUserPostgresDal(_settings);
+            for (var i = 0; i < 10; i++)
+            {
+                var acct = $"myid{i}";
+                var lastTweetId = 1548L;
+
+                await dal.CreateTwitterUserAsync(acct, lastTweetId);
+
+                if (i == 0 || i == 2 || i == 3)
+                {
+                    var t = await dal.GetTwitterUserAsync(acct);
+                    await dal.UpdateTwitterUserAsync(t.Id ,1L,2L, 50+i*2, DateTime.Now);
+                }
+            }
+
+            var result = await dal.GetFailingTwitterUsersCountAsync();
+            Assert.AreEqual(3, result);
+        }
     }
 }

src/Tests/BirdsiteLive.Pipeline.Tests/Processors/RefreshTwitterUserStatusProcessorTests.cs

@@ -0,0 +1,333 @@
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using BirdsiteLive.Common.Settings;
+using BirdsiteLive.DAL.Contracts;
+using BirdsiteLive.DAL.Models;
+using BirdsiteLive.Moderation.Actions;
+using BirdsiteLive.Pipeline.Models;
+using BirdsiteLive.Pipeline.Processors;
+using BirdsiteLive.Twitter;
+using BirdsiteLive.Twitter.Models;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using Moq;
+
+namespace BirdsiteLive.Pipeline.Tests.Processors
+{
+    [TestClass]
+    public class RefreshTwitterUserStatusProcessorTests
+    {
+        [TestMethod]
+        public async Task ProcessAsync_Test()
+        {
+            #region Stubs
+            var userId1 = 1;
+            var userId2 = 2;
+
+            var users = new List<SyncTwitterUser>
+            {
+                new SyncTwitterUser
+                {
+                    Id = userId1
+                },
+                new SyncTwitterUser
+                {
+                    Id = userId2
+                }
+            };
+
+            var settings = new InstanceSettings
+            {
+                FailingTwitterUserCleanUpThreshold = 300
+            };
+            #endregion
+
+            #region Mocks
+            var twitterUserServiceMock = new Mock<ICachedTwitterUserService>(MockBehavior.Strict);
+            twitterUserServiceMock
+                .Setup(x => x.GetUser(It.IsAny<string>()))
+                .Returns(new TwitterUser
+                {
+                    Protected = false
+                });
+
+            var twitterUserDalMock = new Mock<ITwitterUserDal>(MockBehavior.Strict);
+            var removeTwitterAccountActionMock = new Mock<IRemoveTwitterAccountAction>(MockBehavior.Strict);
+            #endregion
+
+            var processor = new RefreshTwitterUserStatusProcessor(twitterUserServiceMock.Object, twitterUserDalMock.Object, removeTwitterAccountActionMock.Object, settings);
+            var result = await processor.ProcessAsync(users.ToArray(), CancellationToken.None);
+
+            #region Validations
+            Assert.AreEqual(2 , result.Length);
+            Assert.IsTrue(result.Any(x => x.User.Id == userId1));
+            Assert.IsTrue(result.Any(x => x.User.Id == userId2));
+
+            twitterUserServiceMock.VerifyAll();
+            twitterUserDalMock.VerifyAll();
+            removeTwitterAccountActionMock.VerifyAll();
+            #endregion
+        }
+
+        [TestMethod]
+        public async Task ProcessAsync_ResetErrorCount_Test()
+        {
+            #region Stubs
+            var userId1 = 1;
+
+            var users = new List<SyncTwitterUser>
+            {
+                new SyncTwitterUser
+                {
+                    Id = userId1,
+                    FetchingErrorCount = 100
+                }
+            };
+
+            var settings = new InstanceSettings
+            {
+                FailingTwitterUserCleanUpThreshold = 300
+            };
+            #endregion
+
+            #region Mocks
+            var twitterUserServiceMock = new Mock<ICachedTwitterUserService>(MockBehavior.Strict);
+            twitterUserServiceMock
+                .Setup(x => x.GetUser(It.IsAny<string>()))
+                .Returns(new TwitterUser
+                {
+                    Protected = false
+                });
+
+            var twitterUserDalMock = new Mock<ITwitterUserDal>(MockBehavior.Strict);
+            var removeTwitterAccountActionMock = new Mock<IRemoveTwitterAccountAction>(MockBehavior.Strict);
+            #endregion
+
+            var processor = new RefreshTwitterUserStatusProcessor(twitterUserServiceMock.Object, twitterUserDalMock.Object, removeTwitterAccountActionMock.Object, settings);
+            var result = await processor.ProcessAsync(users.ToArray(), CancellationToken.None);
+
+            #region Validations
+            Assert.AreEqual(1, result.Length);
+            Assert.IsTrue(result.Any(x => x.User.Id == userId1));
+            Assert.AreEqual(0, result.First().User.FetchingErrorCount);
+
+            twitterUserServiceMock.VerifyAll();
+            twitterUserDalMock.VerifyAll();
+            removeTwitterAccountActionMock.VerifyAll();
+            #endregion
+        }
+
+        [TestMethod]
+        public async Task ProcessAsync_Unfound_Test()
+        {
+            #region Stubs
+            var userId1 = 1;
+            var acct1 = "user1";
+
+            var userId2 = 2;
+            var acct2 = "user2";
+
+            var users = new List<SyncTwitterUser>
+            {
+                new SyncTwitterUser
+                {
+                    Id = userId1,
+                    Acct = acct1
+                },
+                new SyncTwitterUser
+                {
+                    Id = userId2,
+                    Acct = acct2
+                }
+            };
+
+            var settings = new InstanceSettings
+            {
+                FailingTwitterUserCleanUpThreshold = 300
+            };
+            #endregion
+
+            #region Mocks
+            var twitterUserServiceMock = new Mock<ICachedTwitterUserService>(MockBehavior.Strict);
+            twitterUserServiceMock
+                .Setup(x => x.GetUser(It.Is<string>(y => y == acct1)))
+                .Returns(new TwitterUser
+                {
+                    Protected = false
+                });
+
+            twitterUserServiceMock
+                .Setup(x => x.GetUser(It.Is<string>(y => y == acct2)))
+                .Returns((TwitterUser) null);
+
+            twitterUserServiceMock
+                .Setup(x => x.PurgeUser(It.Is<string>(y => y == acct2)));
+
+            var twitterUserDalMock = new Mock<ITwitterUserDal>(MockBehavior.Strict);
+            twitterUserDalMock
+                .Setup(x => x.GetTwitterUserAsync(It.Is<string>(y => y == acct2)))
+                .ReturnsAsync(new SyncTwitterUser
+                {
+                    Id = userId2,
+                    FetchingErrorCount = 0
+                });
+
+            twitterUserDalMock
+                .Setup(x => x.UpdateTwitterUserAsync(It.Is<SyncTwitterUser>(y => y.Id == userId2 && y.FetchingErrorCount == 1)))
+                .Returns(Task.CompletedTask);
+
+            var removeTwitterAccountActionMock = new Mock<IRemoveTwitterAccountAction>(MockBehavior.Strict);
+            #endregion
+
+            var processor = new RefreshTwitterUserStatusProcessor(twitterUserServiceMock.Object, twitterUserDalMock.Object, removeTwitterAccountActionMock.Object, settings);
+            var result = await processor.ProcessAsync(users.ToArray(), CancellationToken.None);
+
+            #region Validations
+            Assert.AreEqual(1, result.Length);
+            Assert.IsTrue(result.Any(x => x.User.Id == userId1));
+
+            twitterUserServiceMock.VerifyAll();
+            twitterUserDalMock.VerifyAll();
+            removeTwitterAccountActionMock.VerifyAll();
+            #endregion
+        }
+
+        [TestMethod]
+        public async Task ProcessAsync_Unfound_OverThreshold_Test()
+        {
+            #region Stubs
+            var userId1 = 1;
+            var acct1 = "user1";
+
+            var userId2 = 2;
+            var acct2 = "user2";
+
+            var users = new List<SyncTwitterUser>
+            {
+                new SyncTwitterUser
+                {
+                    Id = userId1,
+                    Acct = acct1
+                },
+                new SyncTwitterUser
+                {
+                    Id = userId2,
+                    Acct = acct2
+                }
+            };
+
+            var settings = new InstanceSettings
+            {
+                FailingTwitterUserCleanUpThreshold = 300
+            };
+            #endregion
+
+            #region Mocks
+            var twitterUserServiceMock = new Mock<ICachedTwitterUserService>(MockBehavior.Strict);
+            twitterUserServiceMock
+                .Setup(x => x.GetUser(It.Is<string>(y => y == acct1)))
+                .Returns(new TwitterUser
+                {
+                    Protected = false
+                });
+
+            twitterUserServiceMock
+                .Setup(x => x.GetUser(It.Is<string>(y => y == acct2)))
+                .Returns((TwitterUser)null);
+
+            twitterUserServiceMock
+                .Setup(x => x.PurgeUser(It.Is<string>(y => y == acct2)));
+
+            var twitterUserDalMock = new Mock<ITwitterUserDal>(MockBehavior.Strict);
+            twitterUserDalMock
+                .Setup(x => x.GetTwitterUserAsync(It.Is<string>(y => y == acct2)))
+                .ReturnsAsync(new SyncTwitterUser
+                {
+                    Id = userId2,
+                    FetchingErrorCount = 500
+                });
+
+            var removeTwitterAccountActionMock = new Mock<IRemoveTwitterAccountAction>(MockBehavior.Strict);
+            removeTwitterAccountActionMock
+                .Setup(x => x.ProcessAsync(It.Is<SyncTwitterUser>(y => y.Id == userId2)))
+                .Returns(Task.CompletedTask);
+            #endregion
+
+            var processor = new RefreshTwitterUserStatusProcessor(twitterUserServiceMock.Object, twitterUserDalMock.Object, removeTwitterAccountActionMock.Object, settings);
+            var result = await processor.ProcessAsync(users.ToArray(), CancellationToken.None);
+
+            #region Validations
+            Assert.AreEqual(1, result.Length);
+            Assert.IsTrue(result.Any(x => x.User.Id == userId1));
+
+            twitterUserServiceMock.VerifyAll();
+            twitterUserDalMock.VerifyAll();
+            removeTwitterAccountActionMock.VerifyAll();
+            #endregion
+        }
+
+        [TestMethod]
+        public async Task ProcessAsync_Protected_Test()
+        {
+            #region Stubs
+            var userId1 = 1;
+            var acct1 = "user1";
+
+            var userId2 = 2;
+            var acct2 = "user2";
+
+            var users = new List<SyncTwitterUser>
+            {
+                new SyncTwitterUser
+                {
+                    Id = userId1,
+                    Acct = acct1
+                },
+                new SyncTwitterUser
+                {
+                    Id = userId2,
+                    Acct = acct2
+                }
+            };
+
+            var settings = new InstanceSettings
+            {
+                FailingTwitterUserCleanUpThreshold = 300
+            };
+            #endregion
+
+            #region Mocks
+            var twitterUserServiceMock = new Mock<ICachedTwitterUserService>(MockBehavior.Strict);
+            twitterUserServiceMock
+                .Setup(x => x.GetUser(It.Is<string>(y => y == acct1)))
+                .Returns(new TwitterUser
+                {
+                    Protected = false
+                });
+
+            twitterUserServiceMock
+                .Setup(x => x.GetUser(It.Is<string>(y => y == acct2)))
+                .Returns(new TwitterUser
+                {
+                    Protected = true
+                });
+
+            var twitterUserDalMock = new Mock<ITwitterUserDal>(MockBehavior.Strict);
+            var removeTwitterAccountActionMock = new Mock<IRemoveTwitterAccountAction>(MockBehavior.Strict);
+            #endregion
+
+            var processor = new RefreshTwitterUserStatusProcessor(twitterUserServiceMock.Object, twitterUserDalMock.Object, removeTwitterAccountActionMock.Object, settings);
+            var result = await processor.ProcessAsync(users.ToArray(), CancellationToken.None);
+
+            #region Validations
+            Assert.AreEqual(1, result.Length);
+            Assert.IsTrue(result.Any(x => x.User.Id == userId1));
+
+            twitterUserServiceMock.VerifyAll();
+            twitterUserDalMock.VerifyAll();
+            removeTwitterAccountActionMock.VerifyAll();
+            #endregion
+        }
+    }
+}

src/Tests/BirdsiteLive.Pipeline.Tests/Processors/RetrieveFollowersProcessorTests.cs

@@ -21,16 +21,16 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
             var userId1 = 1;
             var userId2 = 2;
 
-            var users = new List<UserWithTweetsToSync>
+            var users = new List<UserWithDataToSync>
             {
-                new UserWithTweetsToSync
+                new UserWithDataToSync
                 {
                     User = new SyncTwitterUser
                     {
                         Id = userId1
                     }
                 },
-                new UserWithTweetsToSync
+                new UserWithDataToSync
                 {
                     User = new SyncTwitterUser
                     {

src/Tests/BirdsiteLive.Pipeline.Tests/Processors/RetrieveTweetsProcessorTests.cs

@@ -4,6 +4,7 @@ using System.Threading;
 using System.Threading.Tasks;
 using BirdsiteLive.DAL.Contracts;
 using BirdsiteLive.DAL.Models;
+using BirdsiteLive.Pipeline.Models;
 using BirdsiteLive.Pipeline.Processors;
 using BirdsiteLive.Twitter;
 using BirdsiteLive.Twitter.Models;
@@ -27,9 +28,14 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
                 LastTweetPostedId = -1
             };
 
+            var user1WtData = new UserWithDataToSync
+            {
+                User = user1,
+            };
+
             var users = new[]
             {
-                user1
+                user1WtData
             };
 
             var tweets = new[]
@@ -57,14 +63,12 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
                     It.Is<int>(y => y == user1.Id),
                     It.Is<long>(y => y == tweets.Last().Id),
                     It.Is<long>(y => y == tweets.Last().Id),
+                    It.Is<int>(y => y == 0),
                     It.IsAny<DateTime>()
                 ))
                 .Returns(Task.CompletedTask);
 
             var twitterUserServiceMock = new Mock<ICachedTwitterUserService>(MockBehavior.Strict);
-            twitterUserServiceMock
-                .Setup(x => x.GetUser(It.Is<string>(y => y == user1.Acct)))
-                .Returns(new TwitterUser {Protected = false});
 
             var logger = new Mock<ILogger<RetrieveTweetsProcessor>>(MockBehavior.Strict);
             #endregion
@@ -94,9 +98,14 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
                 LastTweetSynchronizedForAllFollowersId = 46
             };
 
+            var user1WtData = new UserWithDataToSync
+            {
+                User = user1,
+            };
+
             var users = new[]
             {
-                user1
+                user1WtData
             };
 
             var tweets = new[]
@@ -129,9 +138,6 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
             var twitterUserDalMock = new Mock<ITwitterUserDal>(MockBehavior.Strict);
 
             var twitterUserServiceMock = new Mock<ICachedTwitterUserService>(MockBehavior.Strict);
-            twitterUserServiceMock
-                .Setup(x => x.GetUser(It.Is<string>(y => y == user1.Acct)))
-                .Returns(new TwitterUser { Protected = false });
 
             var logger = new Mock<ILogger<RetrieveTweetsProcessor>>(MockBehavior.Strict);
             #endregion
@@ -147,7 +153,7 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
 
 
             Assert.AreEqual(users.Length, usersResult.Length);
-            Assert.AreEqual(users[0].Acct, usersResult[0].User.Acct);
+            Assert.AreEqual(users[0].User.Acct, usersResult[0].User.Acct);
             Assert.AreEqual(tweets.Length, usersResult[0].Tweets.Length);
             #endregion
         }
@@ -164,9 +170,14 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
                 LastTweetSynchronizedForAllFollowersId = 46
             };
 
+            var user1WtData = new UserWithDataToSync
+            {
+                User = user1,
+            };
+
             var users = new[]
             {
-                user1
+                user1WtData
             };
 
             var tweets = new[]
@@ -199,9 +210,6 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
             var twitterUserDalMock = new Mock<ITwitterUserDal>(MockBehavior.Strict);
 
             var twitterUserServiceMock = new Mock<ICachedTwitterUserService>(MockBehavior.Strict);
-            twitterUserServiceMock
-                .Setup(x => x.GetUser(It.Is<string>(y => y == user1.Acct)))
-                .Returns(new TwitterUser { Protected = false });
 
             var logger = new Mock<ILogger<RetrieveTweetsProcessor>>(MockBehavior.Strict);
             #endregion
@@ -216,7 +224,7 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
             logger.VerifyAll();
 
             Assert.AreEqual(users.Length, usersResult.Length);
-            Assert.AreEqual(users[0].Acct, usersResult[0].User.Acct);
+            Assert.AreEqual(users[0].User.Acct, usersResult[0].User.Acct);
             Assert.AreEqual(tweets.Length, usersResult[0].Tweets.Length);
             #endregion
         }

src/Tests/BirdsiteLive.Pipeline.Tests/Processors/SaveProgressionProcessorTests.cs

@@ -41,7 +41,7 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
                 }
             };
 
-            var usersWithTweets = new UserWithTweetsToSync
+            var usersWithTweets = new UserWithDataToSync
             {
                 Tweets = new []
                 {
@@ -65,6 +65,7 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
                     It.Is<int>(y => y == user.Id),
                     It.Is<long>(y => y == tweet2.Id),
                     It.Is<long>(y => y == tweet2.Id),
+                    It.Is<int>(y => y == 0),
                     It.IsAny<DateTime>()
                 ))
                 .Returns(Task.CompletedTask);
@@ -107,7 +108,7 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
                 }
             };
 
-            var usersWithTweets = new UserWithTweetsToSync
+            var usersWithTweets = new UserWithDataToSync
             {
                 Tweets = new[]
                 {
@@ -130,6 +131,7 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
                     It.Is<int>(y => y == user.Id),
                     It.Is<long>(y => y == tweet3.Id),
                     It.Is<long>(y => y == tweet2.Id),
+                    It.Is<int>(y => y == 0),
                     It.IsAny<DateTime>()
                 ))
                 .Returns(Task.CompletedTask);
@@ -181,7 +183,7 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
                 }
             };
 
-            var usersWithTweets = new UserWithTweetsToSync
+            var usersWithTweets = new UserWithDataToSync
             {
                 Tweets = new[]
                 {
@@ -205,6 +207,7 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
                     It.Is<int>(y => y == user.Id),
                     It.Is<long>(y => y == tweet3.Id),
                     It.Is<long>(y => y == tweet2.Id),
+                    It.Is<int>(y => y == 0),
                     It.IsAny<DateTime>()
                 ))
                 .Returns(Task.CompletedTask);

src/Tests/BirdsiteLive.Pipeline.Tests/Processors/SendTweetsToFollowersProcessorTests.cs

@@ -1,6 +1,8 @@
 using System;
 using System.Threading;
 using System.Threading.Tasks;
+using System.Xml;
+using BirdsiteLive.DAL.Contracts;
 using BirdsiteLive.DAL.Models;
 using BirdsiteLive.Pipeline.Models;
 using BirdsiteLive.Pipeline.Processors;
@@ -26,7 +28,7 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
             var userId2 = 3;
             var userAcct = "user";
 
-            var userWithTweets = new UserWithTweetsToSync()
+            var userWithTweets = new UserWithDataToSync()
             {
                 Tweets = new []
                 {
@@ -69,15 +71,18 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
                     It.Is<Follower[]>(y => y.Length == 2)))
                 .Returns(Task.CompletedTask);
 
+            var followersDalMock = new Mock<IFollowersDal>(MockBehavior.Strict);
+
             var loggerMock = new Mock<ILogger<SendTweetsToFollowersProcessor>>();
             #endregion
 
-            var processor = new SendTweetsToFollowersProcessor(sendTweetsToInboxTaskMock.Object, sendTweetsToSharedInboxTaskMock.Object, loggerMock.Object);
+            var processor = new SendTweetsToFollowersProcessor(sendTweetsToInboxTaskMock.Object, sendTweetsToSharedInboxTaskMock.Object, followersDalMock.Object, loggerMock.Object);
             var result = await processor.ProcessAsync(userWithTweets, CancellationToken.None);
 
             #region Validations
             sendTweetsToInboxTaskMock.VerifyAll();
             sendTweetsToSharedInboxTaskMock.VerifyAll();
+            followersDalMock.VerifyAll();
             #endregion
         }
 
@@ -93,7 +98,7 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
             var userId2 = 3;
             var userAcct = "user";
 
-            var userWithTweets = new UserWithTweetsToSync()
+            var userWithTweets = new UserWithDataToSync()
             {
                 Tweets = new[]
                 {
@@ -139,15 +144,18 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
                     .Returns(Task.CompletedTask);
             }
 
+            var followersDalMock = new Mock<IFollowersDal>(MockBehavior.Strict);
+            
             var loggerMock = new Mock<ILogger<SendTweetsToFollowersProcessor>>();
             #endregion
 
-            var processor = new SendTweetsToFollowersProcessor(sendTweetsToInboxTaskMock.Object, sendTweetsToSharedInboxTaskMock.Object, loggerMock.Object);
+            var processor = new SendTweetsToFollowersProcessor(sendTweetsToInboxTaskMock.Object, sendTweetsToSharedInboxTaskMock.Object, followersDalMock.Object, loggerMock.Object);
             var result = await processor.ProcessAsync(userWithTweets, CancellationToken.None);
 
             #region Validations
             sendTweetsToInboxTaskMock.VerifyAll();
             sendTweetsToSharedInboxTaskMock.VerifyAll();
+            followersDalMock.VerifyAll();
             #endregion
         }
 
@@ -163,7 +171,7 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
             var userId2 = 3;
             var userAcct = "user";
 
-            var userWithTweets = new UserWithTweetsToSync()
+            var userWithTweets = new UserWithDataToSync()
             {
                 Tweets = new[]
                 {
@@ -214,15 +222,193 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
                     It.Is<Follower[]>(y => y.Length == 1)))
                 .Throws(new Exception());
 
+            var followersDalMock = new Mock<IFollowersDal>(MockBehavior.Strict);
+            
+            followersDalMock
+                .Setup(x => x.UpdateFollowerAsync(It.Is<Follower>(y => y.Id == userId2 && y.PostingErrorCount == 1)))
+                .Returns(Task.CompletedTask);
+
             var loggerMock = new Mock<ILogger<SendTweetsToFollowersProcessor>>();
             #endregion
 
-            var processor = new SendTweetsToFollowersProcessor(sendTweetsToInboxTaskMock.Object, sendTweetsToSharedInboxTaskMock.Object, loggerMock.Object);
+            var processor = new SendTweetsToFollowersProcessor(sendTweetsToInboxTaskMock.Object, sendTweetsToSharedInboxTaskMock.Object, followersDalMock.Object, loggerMock.Object);
             var result = await processor.ProcessAsync(userWithTweets, CancellationToken.None);
 
             #region Validations
             sendTweetsToInboxTaskMock.VerifyAll();
             sendTweetsToSharedInboxTaskMock.VerifyAll();
+            followersDalMock.VerifyAll();
+            #endregion
+        }
+
+        [TestMethod]
+        public async Task ProcessAsync_MultiInstances_SharedInbox_OneTweet_ErrorReset_Test()
+        {
+            #region Stubs
+            var tweetId = 1;
+            var host1 = "domain1.ext";
+            var host2 = "domain2.ext";
+            var sharedInbox = "/inbox";
+            var userId1 = 2;
+            var userId2 = 3;
+            var userAcct = "user";
+
+            var userWithTweets = new UserWithDataToSync()
+            {
+                Tweets = new[]
+                {
+                    new ExtractedTweet
+                    {
+                        Id = tweetId
+                    }
+                },
+                User = new SyncTwitterUser
+                {
+                    Acct = userAcct
+                },
+                Followers = new[]
+                {
+                    new Follower
+                    {
+                        Id = userId1,
+                        Host = host1,
+                        SharedInboxRoute = sharedInbox
+                    },
+                    new Follower
+                    {
+                        Id = userId2,
+                        Host = host2,
+                        SharedInboxRoute = sharedInbox,
+                        PostingErrorCount = 50
+                    },
+                }
+            };
+            #endregion
+
+            #region Mocks
+            var sendTweetsToInboxTaskMock = new Mock<ISendTweetsToInboxTask>(MockBehavior.Strict);
+
+            var sendTweetsToSharedInboxTaskMock = new Mock<ISendTweetsToSharedInboxTask>(MockBehavior.Strict);
+            sendTweetsToSharedInboxTaskMock
+                .Setup(x => x.ExecuteAsync(
+                    It.Is<ExtractedTweet[]>(y => y.Length == 1),
+                    It.Is<SyncTwitterUser>(y => y.Acct == userAcct),
+                    It.Is<string>(y => y == host1),
+                    It.Is<Follower[]>(y => y.Length == 1)))
+                .Returns(Task.CompletedTask);
+
+            sendTweetsToSharedInboxTaskMock
+                .Setup(x => x.ExecuteAsync(
+                    It.Is<ExtractedTweet[]>(y => y.Length == 1),
+                    It.Is<SyncTwitterUser>(y => y.Acct == userAcct),
+                    It.Is<string>(y => y == host2),
+                    It.Is<Follower[]>(y => y.Length == 1)))
+                .Returns(Task.CompletedTask);
+
+            var followersDalMock = new Mock<IFollowersDal>(MockBehavior.Strict);
+
+            followersDalMock
+                .Setup(x => x.UpdateFollowerAsync(It.Is<Follower>(y => y.Id == userId2 && y.PostingErrorCount == 0)))
+                .Returns(Task.CompletedTask);
+
+            var loggerMock = new Mock<ILogger<SendTweetsToFollowersProcessor>>();
+            #endregion
+
+            var processor = new SendTweetsToFollowersProcessor(sendTweetsToInboxTaskMock.Object, sendTweetsToSharedInboxTaskMock.Object, followersDalMock.Object, loggerMock.Object);
+            var result = await processor.ProcessAsync(userWithTweets, CancellationToken.None);
+
+            #region Validations
+            sendTweetsToInboxTaskMock.VerifyAll();
+            sendTweetsToSharedInboxTaskMock.VerifyAll();
+            followersDalMock.VerifyAll();
+            #endregion
+        }
+
+        [TestMethod]
+        public async Task ProcessAsync_MultiInstances_SharedInbox_OneTweet_ErrorAndReset_Test()
+        {
+            #region Stubs
+            var tweetId = 1;
+            var host1 = "domain1.ext";
+            var host2 = "domain2.ext";
+            var sharedInbox = "/inbox";
+            var userId1 = 2;
+            var userId2 = 3;
+            var userAcct = "user";
+
+            var userWithTweets = new UserWithDataToSync()
+            {
+                Tweets = new[]
+                {
+                    new ExtractedTweet
+                    {
+                        Id = tweetId
+                    }
+                },
+                User = new SyncTwitterUser
+                {
+                    Acct = userAcct
+                },
+                Followers = new[]
+                {
+                    new Follower
+                    {
+                        Id = userId1,
+                        Host = host1,
+                        SharedInboxRoute = sharedInbox,
+                        PostingErrorCount = 50
+                    },
+                    new Follower
+                    {
+                        Id = userId2,
+                        Host = host2,
+                        SharedInboxRoute = sharedInbox,
+                        PostingErrorCount = 50
+                    },
+                }
+            };
+            #endregion
+
+            #region Mocks
+            var sendTweetsToInboxTaskMock = new Mock<ISendTweetsToInboxTask>(MockBehavior.Strict);
+
+            var sendTweetsToSharedInboxTaskMock = new Mock<ISendTweetsToSharedInboxTask>(MockBehavior.Strict);
+            sendTweetsToSharedInboxTaskMock
+                .Setup(x => x.ExecuteAsync(
+                    It.Is<ExtractedTweet[]>(y => y.Length == 1),
+                    It.Is<SyncTwitterUser>(y => y.Acct == userAcct),
+                    It.Is<string>(y => y == host1),
+                    It.Is<Follower[]>(y => y.Length == 1)))
+                .Returns(Task.CompletedTask);
+
+            sendTweetsToSharedInboxTaskMock
+                .Setup(x => x.ExecuteAsync(
+                    It.Is<ExtractedTweet[]>(y => y.Length == 1),
+                    It.Is<SyncTwitterUser>(y => y.Acct == userAcct),
+                    It.Is<string>(y => y == host2),
+                    It.Is<Follower[]>(y => y.Length == 1)))
+                .Throws(new Exception());
+
+            var followersDalMock = new Mock<IFollowersDal>(MockBehavior.Strict);
+
+            followersDalMock
+                .Setup(x => x.UpdateFollowerAsync(It.Is<Follower>(y => y.Id == userId1 && y.PostingErrorCount == 0)))
+                .Returns(Task.CompletedTask);
+
+            followersDalMock
+                .Setup(x => x.UpdateFollowerAsync(It.Is<Follower>(y => y.Id == userId2 && y.PostingErrorCount == 51)))
+                .Returns(Task.CompletedTask);
+
+            var loggerMock = new Mock<ILogger<SendTweetsToFollowersProcessor>>();
+            #endregion
+
+            var processor = new SendTweetsToFollowersProcessor(sendTweetsToInboxTaskMock.Object, sendTweetsToSharedInboxTaskMock.Object, followersDalMock.Object, loggerMock.Object);
+            var result = await processor.ProcessAsync(userWithTweets, CancellationToken.None);
+
+            #region Validations
+            sendTweetsToInboxTaskMock.VerifyAll();
+            sendTweetsToSharedInboxTaskMock.VerifyAll();
+            followersDalMock.VerifyAll();
             #endregion
         }
 
@@ -237,7 +423,7 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
             var userId2 = 3;
             var userAcct = "user";
 
-            var userWithTweets = new UserWithTweetsToSync()
+            var userWithTweets = new UserWithDataToSync()
             {
                 Tweets = new[]
                 {
@@ -282,15 +468,18 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
 
             var sendTweetsToSharedInboxTaskMock = new Mock<ISendTweetsToSharedInboxTask>(MockBehavior.Strict);
 
+            var followersDalMock = new Mock<IFollowersDal>(MockBehavior.Strict);
+
             var loggerMock = new Mock<ILogger<SendTweetsToFollowersProcessor>>();
             #endregion
 
-            var processor = new SendTweetsToFollowersProcessor(sendTweetsToInboxTaskMock.Object, sendTweetsToSharedInboxTaskMock.Object, loggerMock.Object);
+            var processor = new SendTweetsToFollowersProcessor(sendTweetsToInboxTaskMock.Object, sendTweetsToSharedInboxTaskMock.Object, followersDalMock.Object, loggerMock.Object);
             var result = await processor.ProcessAsync(userWithTweets, CancellationToken.None);
 
             #region Validations
             sendTweetsToInboxTaskMock.VerifyAll();
             sendTweetsToSharedInboxTaskMock.VerifyAll();
+            followersDalMock.VerifyAll();
             #endregion
         }
 
@@ -306,7 +495,7 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
             var userId2 = 3;
             var userAcct = "user";
 
-            var userWithTweets = new UserWithTweetsToSync()
+            var userWithTweets = new UserWithDataToSync()
             {
                 Tweets = new[]
                 {
@@ -351,15 +540,18 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
 
             var sendTweetsToSharedInboxTaskMock = new Mock<ISendTweetsToSharedInboxTask>(MockBehavior.Strict);
 
+            var followersDalMock = new Mock<IFollowersDal>(MockBehavior.Strict);
+
             var loggerMock = new Mock<ILogger<SendTweetsToFollowersProcessor>>();
             #endregion
 
-            var processor = new SendTweetsToFollowersProcessor(sendTweetsToInboxTaskMock.Object, sendTweetsToSharedInboxTaskMock.Object, loggerMock.Object);
+            var processor = new SendTweetsToFollowersProcessor(sendTweetsToInboxTaskMock.Object, sendTweetsToSharedInboxTaskMock.Object, followersDalMock.Object, loggerMock.Object);
             var result = await processor.ProcessAsync(userWithTweets, CancellationToken.None);
 
             #region Validations
             sendTweetsToInboxTaskMock.VerifyAll();
             sendTweetsToSharedInboxTaskMock.VerifyAll();
+            followersDalMock.VerifyAll();
             #endregion
         }
 
@@ -375,7 +567,7 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
             var userId2 = 3;
             var userAcct = "user";
 
-            var userWithTweets = new UserWithTweetsToSync()
+            var userWithTweets = new UserWithDataToSync()
             {
                 Tweets = new[]
                 {
@@ -424,15 +616,189 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
 
             var sendTweetsToSharedInboxTaskMock = new Mock<ISendTweetsToSharedInboxTask>(MockBehavior.Strict);
 
+            var followersDalMock = new Mock<IFollowersDal>(MockBehavior.Strict);
+            
+            followersDalMock
+                .Setup(x => x.UpdateFollowerAsync(It.Is<Follower>(y => y.Id == userId2 && y.PostingErrorCount == 1)))
+                .Returns(Task.CompletedTask);
+
             var loggerMock = new Mock<ILogger<SendTweetsToFollowersProcessor>>();
             #endregion
 
-            var processor = new SendTweetsToFollowersProcessor(sendTweetsToInboxTaskMock.Object, sendTweetsToSharedInboxTaskMock.Object, loggerMock.Object);
+            var processor = new SendTweetsToFollowersProcessor(sendTweetsToInboxTaskMock.Object, sendTweetsToSharedInboxTaskMock.Object, followersDalMock.Object, loggerMock.Object);
             var result = await processor.ProcessAsync(userWithTweets, CancellationToken.None);
 
             #region Validations
             sendTweetsToInboxTaskMock.VerifyAll();
             sendTweetsToSharedInboxTaskMock.VerifyAll();
+            followersDalMock.VerifyAll();
+            #endregion
+        }
+
+        [TestMethod]
+        public async Task ProcessAsync_MultiInstances_Inbox_OneTweet_ErrorReset_Test()
+        {
+            #region Stubs
+            var tweetId = 1;
+            var host1 = "domain1.ext";
+            var host2 = "domain2.ext";
+            var inbox = "/user/inbox";
+            var userId1 = 2;
+            var userId2 = 3;
+            var userAcct = "user";
+
+            var userWithTweets = new UserWithDataToSync()
+            {
+                Tweets = new[]
+                {
+                    new ExtractedTweet
+                    {
+                        Id = tweetId
+                    }
+                },
+                User = new SyncTwitterUser
+                {
+                    Acct = userAcct
+                },
+                Followers = new[]
+                {
+                    new Follower
+                    {
+                        Id = userId1,
+                        Host = host1,
+                        InboxRoute = inbox
+                    },
+                    new Follower
+                    {
+                        Id = userId2,
+                        Host = host2,
+                        InboxRoute = inbox,
+                        PostingErrorCount = 50
+                    },
+                }
+            };
+            #endregion
+
+            #region Mocks
+            var sendTweetsToInboxTaskMock = new Mock<ISendTweetsToInboxTask>(MockBehavior.Strict);
+            sendTweetsToInboxTaskMock
+                .Setup(x => x.ExecuteAsync(
+                    It.Is<ExtractedTweet[]>(y => y.Length == 1),
+                    It.Is<Follower>(y => y.Id == userId1),
+                    It.Is<SyncTwitterUser>(y => y.Acct == userAcct)))
+                .Returns(Task.CompletedTask);
+
+            sendTweetsToInboxTaskMock
+                .Setup(x => x.ExecuteAsync(
+                    It.Is<ExtractedTweet[]>(y => y.Length == 1),
+                    It.Is<Follower>(y => y.Id == userId2),
+                    It.Is<SyncTwitterUser>(y => y.Acct == userAcct)))
+                .Returns(Task.CompletedTask);
+
+            var sendTweetsToSharedInboxTaskMock = new Mock<ISendTweetsToSharedInboxTask>(MockBehavior.Strict);
+
+            var followersDalMock = new Mock<IFollowersDal>(MockBehavior.Strict);
+
+            followersDalMock
+                .Setup(x => x.UpdateFollowerAsync(It.Is<Follower>(y => y.Id == userId2 && y.PostingErrorCount == 0)))
+                .Returns(Task.CompletedTask);
+
+            var loggerMock = new Mock<ILogger<SendTweetsToFollowersProcessor>>();
+            #endregion
+
+            var processor = new SendTweetsToFollowersProcessor(sendTweetsToInboxTaskMock.Object, sendTweetsToSharedInboxTaskMock.Object, followersDalMock.Object, loggerMock.Object);
+            var result = await processor.ProcessAsync(userWithTweets, CancellationToken.None);
+
+            #region Validations
+            sendTweetsToInboxTaskMock.VerifyAll();
+            sendTweetsToSharedInboxTaskMock.VerifyAll();
+            followersDalMock.VerifyAll();
+            #endregion
+        }
+
+        [TestMethod]
+        public async Task ProcessAsync_MultiInstances_Inbox_OneTweet_ErrorAndReset_Test()
+        {
+            #region Stubs
+            var tweetId = 1;
+            var host1 = "domain1.ext";
+            var host2 = "domain2.ext";
+            var inbox = "/user/inbox";
+            var userId1 = 2;
+            var userId2 = 3;
+            var userAcct = "user";
+
+            var userWithTweets = new UserWithDataToSync()
+            {
+                Tweets = new[]
+                {
+                    new ExtractedTweet
+                    {
+                        Id = tweetId
+                    }
+                },
+                User = new SyncTwitterUser
+                {
+                    Acct = userAcct
+                },
+                Followers = new[]
+                {
+                    new Follower
+                    {
+                        Id = userId1,
+                        Host = host1,
+                        InboxRoute = inbox,
+                        PostingErrorCount = 50
+                    },
+                    new Follower
+                    {
+                        Id = userId2,
+                        Host = host2,
+                        InboxRoute = inbox,
+                        PostingErrorCount = 50
+                    },
+                }
+            };
+            #endregion
+
+            #region Mocks
+            var sendTweetsToInboxTaskMock = new Mock<ISendTweetsToInboxTask>(MockBehavior.Strict);
+            sendTweetsToInboxTaskMock
+                .Setup(x => x.ExecuteAsync(
+                    It.Is<ExtractedTweet[]>(y => y.Length == 1),
+                    It.Is<Follower>(y => y.Id == userId1),
+                    It.Is<SyncTwitterUser>(y => y.Acct == userAcct)))
+                .Returns(Task.CompletedTask);
+
+            sendTweetsToInboxTaskMock
+                .Setup(x => x.ExecuteAsync(
+                    It.Is<ExtractedTweet[]>(y => y.Length == 1),
+                    It.Is<Follower>(y => y.Id == userId2),
+                    It.Is<SyncTwitterUser>(y => y.Acct == userAcct)))
+                .Throws(new Exception());
+
+            var sendTweetsToSharedInboxTaskMock = new Mock<ISendTweetsToSharedInboxTask>(MockBehavior.Strict);
+
+            var followersDalMock = new Mock<IFollowersDal>(MockBehavior.Strict);
+
+            followersDalMock
+                .Setup(x => x.UpdateFollowerAsync(It.Is<Follower>(y => y.Id == userId1 && y.PostingErrorCount == 0)))
+                .Returns(Task.CompletedTask);
+
+            followersDalMock
+                .Setup(x => x.UpdateFollowerAsync(It.Is<Follower>(y => y.Id == userId2 && y.PostingErrorCount == 51)))
+                .Returns(Task.CompletedTask);
+
+            var loggerMock = new Mock<ILogger<SendTweetsToFollowersProcessor>>();
+            #endregion
+
+            var processor = new SendTweetsToFollowersProcessor(sendTweetsToInboxTaskMock.Object, sendTweetsToSharedInboxTaskMock.Object, followersDalMock.Object, loggerMock.Object);
+            var result = await processor.ProcessAsync(userWithTweets, CancellationToken.None);
+
+            #region Validations
+            sendTweetsToInboxTaskMock.VerifyAll();
+            sendTweetsToSharedInboxTaskMock.VerifyAll();
+            followersDalMock.VerifyAll();
             #endregion
         }
     }

src/Tests/BirdsiteLive.Pipeline.Tests/StatusPublicationPipelineTests.cs

@@ -27,6 +27,7 @@ namespace BirdsiteLive.Pipeline.Tests
                     It.IsAny<CancellationToken>()))
                 .Returns(Task.Delay(0));
 
+            var refreshTwitterUserStatusProcessor = new Mock<IRefreshTwitterUserStatusProcessor>(MockBehavior.Strict);
             var retrieveTweetsProcessor = new Mock<IRetrieveTweetsProcessor>(MockBehavior.Strict);
             var retrieveFollowersProcessor = new Mock<IRetrieveFollowersProcessor>(MockBehavior.Strict);
             var sendTweetsToFollowersProcessor = new Mock<ISendTweetsToFollowersProcessor>(MockBehavior.Strict);
@@ -34,7 +35,7 @@ namespace BirdsiteLive.Pipeline.Tests
             var logger = new Mock<ILogger<StatusPublicationPipeline>>();
             #endregion
 
-            var pipeline = new StatusPublicationPipeline(retrieveTweetsProcessor.Object, retrieveTwitterUsersProcessor.Object, retrieveFollowersProcessor.Object, sendTweetsToFollowersProcessor.Object, saveProgressionProcessor.Object, logger.Object);
+            var pipeline = new StatusPublicationPipeline(retrieveTweetsProcessor.Object, retrieveTwitterUsersProcessor.Object, retrieveFollowersProcessor.Object, sendTweetsToFollowersProcessor.Object, saveProgressionProcessor.Object, refreshTwitterUserStatusProcessor.Object, logger.Object);
             await pipeline.ExecuteAsync(ct.Token);
 
             #region Validations