diff --git a/VARIABLES.md b/VARIABLES.md index e17b3a5..e9f6f33 100644 --- a/VARIABLES.md +++ b/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 diff --git a/src/BSLManager/App.cs b/src/BSLManager/App.cs index 0e48262..37697cc 100644 --- a/src/BSLManager/App.cs +++ b/src/BSLManager/App.cs @@ -146,23 +146,31 @@ private void OpenFollowerDialog(int selectedIndex) 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); diff --git a/src/BSLManager/Domain/FollowersListState.cs b/src/BSLManager/Domain/FollowersListState.cs index f33acb8..02c2151 100644 --- a/src/BSLManager/Domain/FollowersListState.cs +++ b/src/BSLManager/Domain/FollowersListState.cs @@ -26,7 +26,7 @@ private void ResetLists() 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); } } diff --git a/src/BirdsiteLive.Common/Settings/InstanceSettings.cs b/src/BirdsiteLive.Common/Settings/InstanceSettings.cs index 9353398..a85d9e6 100644 --- a/src/BirdsiteLive.Common/Settings/InstanceSettings.cs +++ b/src/BirdsiteLive.Common/Settings/InstanceSettings.cs @@ -11,5 +11,7 @@ public class InstanceSettings public string UnlistedTwitterAccounts { get; set; } public string SensitiveTwitterAccounts { get; set; } + + public int FailingTwitterUserCleanUpThreshold { get; set; } } } diff --git a/src/BirdsiteLive.Pipeline/BirdsiteLive.Pipeline.csproj b/src/BirdsiteLive.Pipeline/BirdsiteLive.Pipeline.csproj index 884af18..8601b19 100644 --- a/src/BirdsiteLive.Pipeline/BirdsiteLive.Pipeline.csproj +++ b/src/BirdsiteLive.Pipeline/BirdsiteLive.Pipeline.csproj @@ -13,6 +13,7 @@ + diff --git a/src/BirdsiteLive.Pipeline/Contracts/IRefreshTwitterUserStatusProcessor.cs b/src/BirdsiteLive.Pipeline/Contracts/IRefreshTwitterUserStatusProcessor.cs new file mode 100644 index 0000000..9f20e59 --- /dev/null +++ b/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 ProcessAsync(SyncTwitterUser[] syncTwitterUsers, CancellationToken ct); + } +} \ No newline at end of file diff --git a/src/BirdsiteLive.Pipeline/Contracts/IRetrieveFollowersProcessor.cs b/src/BirdsiteLive.Pipeline/Contracts/IRetrieveFollowersProcessor.cs index e0d45dc..a9ef35c 100644 --- a/src/BirdsiteLive.Pipeline/Contracts/IRetrieveFollowersProcessor.cs +++ b/src/BirdsiteLive.Pipeline/Contracts/IRetrieveFollowersProcessor.cs @@ -7,7 +7,7 @@ namespace BirdsiteLive.Pipeline.Contracts { public interface IRetrieveFollowersProcessor { - Task> ProcessAsync(UserWithTweetsToSync[] userWithTweetsToSyncs, CancellationToken ct); + Task> ProcessAsync(UserWithDataToSync[] userWithTweetsToSyncs, CancellationToken ct); //IAsyncEnumerable ProcessAsync(UserWithTweetsToSync[] userWithTweetsToSyncs, CancellationToken ct); } } \ No newline at end of file diff --git a/src/BirdsiteLive.Pipeline/Contracts/IRetrieveTweetsProcessor.cs b/src/BirdsiteLive.Pipeline/Contracts/IRetrieveTweetsProcessor.cs index 451f1d1..49712c2 100644 --- a/src/BirdsiteLive.Pipeline/Contracts/IRetrieveTweetsProcessor.cs +++ b/src/BirdsiteLive.Pipeline/Contracts/IRetrieveTweetsProcessor.cs @@ -7,6 +7,6 @@ namespace BirdsiteLive.Pipeline.Contracts { public interface IRetrieveTweetsProcessor { - Task ProcessAsync(SyncTwitterUser[] syncTwitterUsers, CancellationToken ct); + Task ProcessAsync(UserWithDataToSync[] syncTwitterUsers, CancellationToken ct); } } \ No newline at end of file diff --git a/src/BirdsiteLive.Pipeline/Contracts/ISaveProgressionProcessor.cs b/src/BirdsiteLive.Pipeline/Contracts/ISaveProgressionProcessor.cs index 02efaef..6b1c9ba 100644 --- a/src/BirdsiteLive.Pipeline/Contracts/ISaveProgressionProcessor.cs +++ b/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); } } \ No newline at end of file diff --git a/src/BirdsiteLive.Pipeline/Contracts/ISendTweetsToFollowersProcessor.cs b/src/BirdsiteLive.Pipeline/Contracts/ISendTweetsToFollowersProcessor.cs index 6d55957..33db423 100644 --- a/src/BirdsiteLive.Pipeline/Contracts/ISendTweetsToFollowersProcessor.cs +++ b/src/BirdsiteLive.Pipeline/Contracts/ISendTweetsToFollowersProcessor.cs @@ -6,6 +6,6 @@ namespace BirdsiteLive.Pipeline.Contracts { public interface ISendTweetsToFollowersProcessor { - Task ProcessAsync(UserWithTweetsToSync userWithTweetsToSync, CancellationToken ct); + Task ProcessAsync(UserWithDataToSync userWithTweetsToSync, CancellationToken ct); } } \ No newline at end of file diff --git a/src/BirdsiteLive.Pipeline/Models/UserWithTweetsToSync.cs b/src/BirdsiteLive.Pipeline/Models/UserWithDataToSync.cs similarity index 88% rename from src/BirdsiteLive.Pipeline/Models/UserWithTweetsToSync.cs rename to src/BirdsiteLive.Pipeline/Models/UserWithDataToSync.cs index 57810c7..c5e6639 100644 --- a/src/BirdsiteLive.Pipeline/Models/UserWithTweetsToSync.cs +++ b/src/BirdsiteLive.Pipeline/Models/UserWithDataToSync.cs @@ -4,7 +4,7 @@ namespace BirdsiteLive.Pipeline.Models { - public class UserWithTweetsToSync + public class UserWithDataToSync { public SyncTwitterUser User { get; set; } public ExtractedTweet[] Tweets { get; set; } diff --git a/src/BirdsiteLive.Pipeline/Processors/RefreshTwitterUserStatusProcessor.cs b/src/BirdsiteLive.Pipeline/Processors/RefreshTwitterUserStatusProcessor.cs new file mode 100644 index 0000000..dae9fe0 --- /dev/null +++ b/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 ProcessAsync(SyncTwitterUser[] syncTwitterUsers, CancellationToken ct) + { + var usersWtData = new List(); + + 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); + } + } +} \ No newline at end of file diff --git a/src/BirdsiteLive.Pipeline/Processors/RetrieveFollowersProcessor.cs b/src/BirdsiteLive.Pipeline/Processors/RetrieveFollowersProcessor.cs index 4b2f150..57e3e49 100644 --- a/src/BirdsiteLive.Pipeline/Processors/RetrieveFollowersProcessor.cs +++ b/src/BirdsiteLive.Pipeline/Processors/RetrieveFollowersProcessor.cs @@ -18,7 +18,7 @@ public RetrieveFollowersProcessor(IFollowersDal followersDal) } #endregion - public async Task> ProcessAsync(UserWithTweetsToSync[] userWithTweetsToSyncs, CancellationToken ct) + public async Task> ProcessAsync(UserWithDataToSync[] userWithTweetsToSyncs, CancellationToken ct) { //TODO multithread this foreach (var user in userWithTweetsToSyncs) diff --git a/src/BirdsiteLive.Pipeline/Processors/RetrieveTweetsProcessor.cs b/src/BirdsiteLive.Pipeline/Processors/RetrieveTweetsProcessor.cs index bb5e026..58d35d0 100644 --- a/src/BirdsiteLive.Pipeline/Processors/RetrieveTweetsProcessor.cs +++ b/src/BirdsiteLive.Pipeline/Processors/RetrieveTweetsProcessor.cs @@ -31,33 +31,30 @@ public RetrieveTweetsProcessor(ITwitterTweetsService twitterTweetsService, ITwit } #endregion - public async Task ProcessAsync(SyncTwitterUser[] syncTwitterUsers, CancellationToken ct) + public async Task ProcessAsync(UserWithDataToSync[] syncTwitterUsers, CancellationToken ct) { - var usersWtTweets = new List(); + var usersWtTweets = new List(); //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 @@ public async Task ProcessAsync(SyncTwitterUser[] syncTwi 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) diff --git a/src/BirdsiteLive.Pipeline/Processors/RetrieveTwitterUsersProcessor.cs b/src/BirdsiteLive.Pipeline/Processors/RetrieveTwitterUsersProcessor.cs index ebb87fc..973b672 100644 --- a/src/BirdsiteLive.Pipeline/Processors/RetrieveTwitterUsersProcessor.cs +++ b/src/BirdsiteLive.Pipeline/Processors/RetrieveTwitterUsersProcessor.cs @@ -55,7 +55,12 @@ public async Task GetTwitterUsersAsync(BufferBlock twitterUse } 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) { diff --git a/src/BirdsiteLive.Pipeline/Processors/SaveProgressionProcessor.cs b/src/BirdsiteLive.Pipeline/Processors/SaveProgressionProcessor.cs index c2f3ff5..1437255 100644 --- a/src/BirdsiteLive.Pipeline/Processors/SaveProgressionProcessor.cs +++ b/src/BirdsiteLive.Pipeline/Processors/SaveProgressionProcessor.cs @@ -22,7 +22,7 @@ public SaveProgressionProcessor(ITwitterUserDal twitterUserDal, ILogger 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) { diff --git a/src/BirdsiteLive.Pipeline/Processors/SendTweetsToFollowersProcessor.cs b/src/BirdsiteLive.Pipeline/Processors/SendTweetsToFollowersProcessor.cs index afdb00e..65f9610 100644 --- a/src/BirdsiteLive.Pipeline/Processors/SendTweetsToFollowersProcessor.cs +++ b/src/BirdsiteLive.Pipeline/Processors/SendTweetsToFollowersProcessor.cs @@ -22,18 +22,20 @@ public class SendTweetsToFollowersProcessor : ISendTweetsToFollowersProcessor { private readonly ISendTweetsToInboxTask _sendTweetsToInboxTask; private readonly ISendTweetsToSharedInboxTask _sendTweetsToSharedInbox; + private readonly IFollowersDal _followersDal; private readonly ILogger _logger; #region Ctor - public SendTweetsToFollowersProcessor(ISendTweetsToInboxTask sendTweetsToInboxTask, ISendTweetsToSharedInboxTask sendTweetsToSharedInbox, ILogger logger) + public SendTweetsToFollowersProcessor(ISendTweetsToInboxTask sendTweetsToInboxTask, ISendTweetsToSharedInboxTask sendTweetsToSharedInbox, IFollowersDal followersDal, ILogger logger) { _sendTweetsToInboxTask = sendTweetsToInboxTask; _sendTweetsToSharedInbox = sendTweetsToSharedInbox; _logger = logger; + _followersDal = followersDal; } #endregion - public async Task ProcessAsync(UserWithTweetsToSync userWithTweetsToSync, CancellationToken ct) + public async Task ProcessAsync(UserWithDataToSync userWithTweetsToSync, CancellationToken ct) { var user = userWithTweetsToSync.User; @@ -41,18 +43,18 @@ public async Task ProcessAsync(UserWithTweetsToSync userWi 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 followers, SyncTwitterUser user) + private async Task ProcessFollowersWithSharedInboxAsync(ExtractedTweet[] tweets, List followers, SyncTwitterUser user) { var followersPerInstances = followers.GroupBy(x => x.Host); @@ -61,28 +63,51 @@ private async Task ProcessFollowersWithSharedInbox(ExtractedTweet[] tweets, List 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 followerWtInbox, SyncTwitterUser user) + private async Task ProcessFollowersWithInboxAsync(ExtractedTweet[] tweets, List 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); + } } } \ No newline at end of file diff --git a/src/BirdsiteLive.Pipeline/StatusPublicationPipeline.cs b/src/BirdsiteLive.Pipeline/StatusPublicationPipeline.cs index d2436f0..c6917e7 100644 --- a/src/BirdsiteLive.Pipeline/StatusPublicationPipeline.cs +++ b/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 @@ public interface IStatusPublicationPipeline 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 @@ public class StatusPublicationPipeline : IStatusPublicationPipeline private readonly ILogger _logger; #region Ctor - public StatusPublicationPipeline(IRetrieveTweetsProcessor retrieveTweetsProcessor, IRetrieveTwitterUsersProcessor retrieveTwitterAccountsProcessor, IRetrieveFollowersProcessor retrieveFollowersProcessor, ISendTweetsToFollowersProcessor sendTweetsToFollowersProcessor, ISaveProgressionProcessor saveProgressionProcessor, ILogger logger) + public StatusPublicationPipeline(IRetrieveTweetsProcessor retrieveTweetsProcessor, IRetrieveTwitterUsersProcessor retrieveTwitterAccountsProcessor, IRetrieveFollowersProcessor retrieveFollowersProcessor, ISendTweetsToFollowersProcessor sendTweetsToFollowersProcessor, ISaveProgressionProcessor saveProgressionProcessor, IRefreshTwitterUserStatusProcessor refreshTwitterUserStatusProcessor, ILogger logger) { _retrieveTweetsProcessor = retrieveTweetsProcessor; _retrieveTwitterAccountsProcessor = retrieveTwitterAccountsProcessor; _retrieveFollowersProcessor = retrieveFollowersProcessor; _sendTweetsToFollowersProcessor = sendTweetsToFollowersProcessor; _saveProgressionProcessor = saveProgressionProcessor; + _refreshTwitterUserStatusProcessor = refreshTwitterUserStatusProcessor; _logger = logger; } @@ -39,16 +42,21 @@ public StatusPublicationPipeline(IRetrieveTweetsProcessor retrieveTweetsProcesso public async Task ExecuteAsync(CancellationToken ct) { // Create blocks - var twitterUsersBufferBlock = new BufferBlock(new DataflowBlockOptions { BoundedCapacity = 1, CancellationToken = ct }); - var retrieveTweetsBlock = new TransformBlock(async x => await _retrieveTweetsProcessor.ProcessAsync(x, ct)); - var retrieveTweetsBufferBlock = new BufferBlock(new DataflowBlockOptions { BoundedCapacity = 1, CancellationToken = ct }); - var retrieveFollowersBlock = new TransformManyBlock(async x => await _retrieveFollowersProcessor.ProcessAsync(x, ct)); - var retrieveFollowersBufferBlock = new BufferBlock(new DataflowBlockOptions { BoundedCapacity = 20, CancellationToken = ct }); - var sendTweetsToFollowersBlock = new TransformBlock(async x => await _sendTweetsToFollowersProcessor.ProcessAsync(x, ct), new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 5, CancellationToken = ct }); - var sendTweetsToFollowersBufferBlock = new BufferBlock(new DataflowBlockOptions { BoundedCapacity = 20, CancellationToken = ct }); - var saveProgressionBlock = new ActionBlock(async x => await _saveProgressionProcessor.ProcessAsync(x, ct), new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 5, CancellationToken = ct }); + var twitterUserToRefreshBufferBlock = new BufferBlock(new DataflowBlockOptions + { BoundedCapacity = 1, CancellationToken = ct }); + var twitterUserToRefreshBlock = new TransformBlock(async x => await _refreshTwitterUserStatusProcessor.ProcessAsync(x, ct)); + var twitterUsersBufferBlock = new BufferBlock(new DataflowBlockOptions { BoundedCapacity = 1, CancellationToken = ct }); + var retrieveTweetsBlock = new TransformBlock(async x => await _retrieveTweetsProcessor.ProcessAsync(x, ct)); + var retrieveTweetsBufferBlock = new BufferBlock(new DataflowBlockOptions { BoundedCapacity = 1, CancellationToken = ct }); + var retrieveFollowersBlock = new TransformManyBlock(async x => await _retrieveFollowersProcessor.ProcessAsync(x, ct)); + var retrieveFollowersBufferBlock = new BufferBlock(new DataflowBlockOptions { BoundedCapacity = 20, CancellationToken = ct }); + var sendTweetsToFollowersBlock = new TransformBlock(async x => await _sendTweetsToFollowersProcessor.ProcessAsync(x, ct), new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 5, CancellationToken = ct }); + var sendTweetsToFollowersBufferBlock = new BufferBlock(new DataflowBlockOptions { BoundedCapacity = 20, CancellationToken = ct }); + var saveProgressionBlock = new ActionBlock(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 @@ public async Task ExecuteAsync(CancellationToken ct) 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 }); diff --git a/src/BirdsiteLive.Twitter/Statistics/TwitterStatisticsHandler.cs b/src/BirdsiteLive.Twitter/Statistics/TwitterStatisticsHandler.cs index 04a5649..668be76 100644 --- a/src/BirdsiteLive.Twitter/Statistics/TwitterStatisticsHandler.cs +++ b/src/BirdsiteLive.Twitter/Statistics/TwitterStatisticsHandler.cs @@ -95,7 +95,7 @@ public ApiStatistics GetStatistics() TimelineCallsCountMin = timelineCalls.Any() ? timelineCalls.Min() : 0, TimelineCallsCountAvg = timelineCalls.Any() ? (int)timelineCalls.Average() : 0, TimelineCallsCountMax = timelineCalls.Any() ? timelineCalls.Max() : 0, - TimelineCallsMax = 1500 + TimelineCallsMax = 1000 }; } } diff --git a/src/BirdsiteLive.Twitter/TwitterUserService.cs b/src/BirdsiteLive.Twitter/TwitterUserService.cs index 2370cea..6a27dc1 100644 --- a/src/BirdsiteLive.Twitter/TwitterUserService.cs +++ b/src/BirdsiteLive.Twitter/TwitterUserService.cs @@ -49,6 +49,9 @@ public TwitterUser GetUser(string username) 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; } diff --git a/src/BirdsiteLive/BirdsiteLive.csproj b/src/BirdsiteLive/BirdsiteLive.csproj index 2711007..25ea8a9 100644 --- a/src/BirdsiteLive/BirdsiteLive.csproj +++ b/src/BirdsiteLive/BirdsiteLive.csproj @@ -4,7 +4,7 @@ netcoreapp3.1 d21486de-a812-47eb-a419-05682bb68856 Linux - 0.18.0 + 0.19.0 diff --git a/src/BirdsiteLive/Controllers/StatisticsController.cs b/src/BirdsiteLive/Controllers/StatisticsController.cs index 5a9aef8..4078a31 100644 --- a/src/BirdsiteLive/Controllers/StatisticsController.cs +++ b/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 @@ public class StatisticsController : Controller private readonly ITwitterStatisticsHandler _twitterStatistics; private readonly IExtractionStatisticsHandler _extractionStatistics; - #region Ctor public StatisticsController(ITwitterUserDal twitterUserDal, IFollowersDal followersDal, ITwitterStatisticsHandler twitterStatistics, IExtractionStatisticsHandler extractionStatistics) { @@ -32,7 +32,9 @@ public async Task Index() 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(), }; diff --git a/src/BirdsiteLive/Controllers/UsersController.cs b/src/BirdsiteLive/Controllers/UsersController.cs index 8240d7b..0d5f77b 100644 --- a/src/BirdsiteLive/Controllers/UsersController.cs +++ b/src/BirdsiteLive/Controllers/UsersController.cs @@ -58,6 +58,7 @@ public IActionResult Index() [Route("/@{id}")] [Route("/users/{id}")] + [Route("/users/{id}/remote_follow")] public IActionResult Index(string id) { _logger.LogTrace("User Index: {Id}", id); diff --git a/src/BirdsiteLive/Models/StatisticsModels/Statistics.cs b/src/BirdsiteLive/Models/StatisticsModels/Statistics.cs index a2dd693..4a66f81 100644 --- a/src/BirdsiteLive/Models/StatisticsModels/Statistics.cs +++ b/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; } } diff --git a/src/BirdsiteLive/Views/Statistics/Index.cshtml b/src/BirdsiteLive/Views/Statistics/Index.cshtml index c3dd57f..4382e2a 100644 --- a/src/BirdsiteLive/Views/Statistics/Index.cshtml +++ b/src/BirdsiteLive/Views/Statistics/Index.cshtml @@ -9,7 +9,9 @@

Instance

  • Twitter Users: @Model.TwitterUserCount
    • +
  • Failing Twitter Users: @Model.FailingTwitterUserCount
  • Followers: @Model.FollowersCount
    • +
  • Failing Followers: @Model.FailingFollowersCount

Twitter API (Min, Avg, Max for the last 24h)

diff --git a/src/BirdsiteLive/appsettings.json b/src/BirdsiteLive/appsettings.json index d37ee5e..d510809 100644 --- a/src/BirdsiteLive/appsettings.json +++ b/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", diff --git a/src/DataAccessLayers/BirdsiteLive.DAL.Postgres/DataAccessLayers/DbInitializerPostgresDal.cs b/src/DataAccessLayers/BirdsiteLive.DAL.Postgres/DataAccessLayers/DbInitializerPostgresDal.cs index 814578e..2e3acea 100644 --- a/src/DataAccessLayers/BirdsiteLive.DAL.Postgres/DataAccessLayers/DbInitializerPostgresDal.cs +++ b/src/DataAccessLayers/BirdsiteLive.DAL.Postgres/DataAccessLayers/DbInitializerPostgresDal.cs @@ -23,7 +23,7 @@ internal class DbVersion 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 @@ public Tuple[] GetMigrationPatterns() return new[] { new Tuple(new Version(1,0), new Version(2,0)), - new Tuple(new Version(2,0), new Version(2,1)) + new Tuple(new Version(2,0), new Version(2,1)), + new Tuple(new Version(2,1), new Version(2,2)), + new Tuple(new Version(2,2), new Version(2,3)) }; } @@ -151,6 +153,16 @@ public async Task MigrateDbAsync(Version from, Version to) 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(); diff --git a/src/DataAccessLayers/BirdsiteLive.DAL.Postgres/DataAccessLayers/FollowersPostgresDal.cs b/src/DataAccessLayers/BirdsiteLive.DAL.Postgres/DataAccessLayers/FollowersPostgresDal.cs index ec031d4..db2f9f7 100644 --- a/src/DataAccessLayers/BirdsiteLive.DAL.Postgres/DataAccessLayers/FollowersPostgresDal.cs +++ b/src/DataAccessLayers/BirdsiteLive.DAL.Postgres/DataAccessLayers/FollowersPostgresDal.cs @@ -53,6 +53,19 @@ public async Task GetFollowersCountAsync() } } + public async Task GetFailingFollowersCountAsync() + { + var query = $"SELECT COUNT(*) FROM {_settings.FollowersTableName} WHERE postingErrorCount > 0"; + + using (var dbConnection = Connection) + { + dbConnection.Open(); + + var result = (await dbConnection.QueryAsync(query)).FirstOrDefault(); + return result; + } + } + public async Task GetFollowerAsync(string acct, string host) { var query = $"SELECT * FROM {_settings.FollowersTableName} WHERE acct = @acct AND host = @host"; @@ -103,13 +116,13 @@ public async Task UpdateFollowerAsync(Follower follower) 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 @@ private Follower Convert(SerializedFollower follower) ActorId = follower.ActorId, SharedInboxRoute = follower.SharedInboxRoute, Followings = follower.Followings.ToList(), - FollowingsSyncStatus = JsonConvert.DeserializeObject>(follower.FollowingsSyncStatus) + FollowingsSyncStatus = JsonConvert.DeserializeObject>(follower.FollowingsSyncStatus), + PostingErrorCount = follower.PostingErrorCount }; } } @@ -174,5 +188,6 @@ internal class SerializedFollower { public string InboxRoute { get; set; } public string SharedInboxRoute { get; set; } public string ActorId { get; set; } + public int PostingErrorCount { get; set; } } } \ No newline at end of file diff --git a/src/DataAccessLayers/BirdsiteLive.DAL.Postgres/DataAccessLayers/TwitterUserPostgresDal.cs b/src/DataAccessLayers/BirdsiteLive.DAL.Postgres/DataAccessLayers/TwitterUserPostgresDal.cs index 855df5e..48b8455 100644 --- a/src/DataAccessLayers/BirdsiteLive.DAL.Postgres/DataAccessLayers/TwitterUserPostgresDal.cs +++ b/src/DataAccessLayers/BirdsiteLive.DAL.Postgres/DataAccessLayers/TwitterUserPostgresDal.cs @@ -73,6 +73,19 @@ public async Task GetTwitterUsersCountAsync() } } + public async Task GetFailingTwitterUsersCountAsync() + { + var query = $"SELECT COUNT(*) FROM {_settings.TwitterUserTableName} WHERE fetchingErrorCount > 0"; + + using (var dbConnection = Connection) + { + dbConnection.Open(); + + var result = (await dbConnection.QueryAsync(query)).FirstOrDefault(); + return result; + } + } + public async Task GetAllTwitterUsersAsync(int maxNumber) { var query = $"SELECT * FROM {_settings.TwitterUserTableName} ORDER BY lastSync ASC LIMIT @maxNumber"; @@ -99,23 +112,28 @@ public async Task GetAllTwitterUsersAsync() } } - 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"); diff --git a/src/DataAccessLayers/BirdsiteLive.DAL/Contracts/IFollowersDal.cs b/src/DataAccessLayers/BirdsiteLive.DAL/Contracts/IFollowersDal.cs index 86caa02..fe87b28 100644 --- a/src/DataAccessLayers/BirdsiteLive.DAL/Contracts/IFollowersDal.cs +++ b/src/DataAccessLayers/BirdsiteLive.DAL/Contracts/IFollowersDal.cs @@ -15,5 +15,6 @@ Task CreateFollowerAsync(string acct, string host, string inboxRoute, string sha Task DeleteFollowerAsync(int id); Task DeleteFollowerAsync(string acct, string host); Task GetFollowersCountAsync(); + Task GetFailingFollowersCountAsync(); } } \ No newline at end of file diff --git a/src/DataAccessLayers/BirdsiteLive.DAL/Contracts/ITwitterUserDal.cs b/src/DataAccessLayers/BirdsiteLive.DAL/Contracts/ITwitterUserDal.cs index cfa422a..ef2cc36 100644 --- a/src/DataAccessLayers/BirdsiteLive.DAL/Contracts/ITwitterUserDal.cs +++ b/src/DataAccessLayers/BirdsiteLive.DAL/Contracts/ITwitterUserDal.cs @@ -11,9 +11,11 @@ public interface ITwitterUserDal Task GetTwitterUserAsync(int id); Task GetAllTwitterUsersAsync(int maxNumber); Task 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 GetTwitterUsersCountAsync(); + Task GetFailingTwitterUsersCountAsync(); } } \ No newline at end of file diff --git a/src/DataAccessLayers/BirdsiteLive.DAL/Models/Follower.cs b/src/DataAccessLayers/BirdsiteLive.DAL/Models/Follower.cs index 274852b..357e32e 100644 --- a/src/DataAccessLayers/BirdsiteLive.DAL/Models/Follower.cs +++ b/src/DataAccessLayers/BirdsiteLive.DAL/Models/Follower.cs @@ -14,5 +14,7 @@ public class Follower public string Host { get; set; } public string InboxRoute { get; set; } public string SharedInboxRoute { get; set; } + + public int PostingErrorCount { get; set; } } } \ No newline at end of file diff --git a/src/DataAccessLayers/BirdsiteLive.DAL/Models/SyncTwitterUser.cs b/src/DataAccessLayers/BirdsiteLive.DAL/Models/SyncTwitterUser.cs index 59be0a5..8b18ba1 100644 --- a/src/DataAccessLayers/BirdsiteLive.DAL/Models/SyncTwitterUser.cs +++ b/src/DataAccessLayers/BirdsiteLive.DAL/Models/SyncTwitterUser.cs @@ -11,5 +11,7 @@ public class SyncTwitterUser public long LastTweetSynchronizedForAllFollowersId { get; set; } public DateTime LastSync { get; set; } + + public int FetchingErrorCount { get; set; } //TODO: update DAL } } \ No newline at end of file diff --git a/src/Tests/BirdsiteLive.DAL.Postgres.Tests/DataAccessLayers/FollowersPostgresDalTests.cs b/src/Tests/BirdsiteLive.DAL.Postgres.Tests/DataAccessLayers/FollowersPostgresDalTests.cs index cbaeb72..a22df0f 100644 --- a/src/Tests/BirdsiteLive.DAL.Postgres.Tests/DataAccessLayers/FollowersPostgresDalTests.cs +++ b/src/Tests/BirdsiteLive.DAL.Postgres.Tests/DataAccessLayers/FollowersPostgresDalTests.cs @@ -54,6 +54,7 @@ public async Task CreateAndGetFollower() 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 @@ public async Task CreateAndGetFollower_NoFollowings() 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 @@ public async Task CreateAndGetFollower_NoSharedInbox() 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 @@ public async Task CountFollowersAsync() 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 @@ public async Task CountFollowersAsync() 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(); + 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 @@ public async Task CreateUpdateAndGetFollower_Add() }; result.Followings = updatedFollowing.ToList(); result.FollowingsSyncStatus = updatedFollowingSync; - - + result.PostingErrorCount = 10; + await dal.UpdateFollowerAsync(result); result = await dal.GetFollowerAsync(acct, host); @@ -286,6 +337,7 @@ public async Task CreateUpdateAndGetFollower_Add() 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 @@ public async Task CreateUpdateAndGetFollower_Remove() }; result.Followings = updatedFollowing.ToList(); result.FollowingsSyncStatus = updatedFollowingSync; + result.PostingErrorCount = 5; await dal.UpdateFollowerAsync(result); result = await dal.GetFollowerAsync(acct, host); @@ -325,6 +378,41 @@ public async Task CreateUpdateAndGetFollower_Remove() 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() + { + {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] diff --git a/src/Tests/BirdsiteLive.DAL.Postgres.Tests/DataAccessLayers/TwitterUserPostgresDalTests.cs b/src/Tests/BirdsiteLive.DAL.Postgres.Tests/DataAccessLayers/TwitterUserPostgresDalTests.cs index 0cf3ca1..68a060f 100644 --- a/src/Tests/BirdsiteLive.DAL.Postgres.Tests/DataAccessLayers/TwitterUserPostgresDalTests.cs +++ b/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 @@ public async Task CreateAndGetUser() 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 @@ public async Task CreateUpdateAndGetUser() 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 @@ public async Task CreateUpdateAndGetUser() 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 @@ public async Task Update_NoId() 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 @@ public async Task Update_NoLastTweetPostedId() 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 @@ public async Task Update_NoLastTweetSynchronizedForAllFollowersId() 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 @@ public async Task GetAllTwitterUsers_Limited() { 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 @@ public async Task CountTwitterUsers() 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); + } } } \ No newline at end of file diff --git a/src/Tests/BirdsiteLive.Pipeline.Tests/Processors/RefreshTwitterUserStatusProcessorTests.cs b/src/Tests/BirdsiteLive.Pipeline.Tests/Processors/RefreshTwitterUserStatusProcessorTests.cs new file mode 100644 index 0000000..52ace24 --- /dev/null +++ b/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 + { + new SyncTwitterUser + { + Id = userId1 + }, + new SyncTwitterUser + { + Id = userId2 + } + }; + + var settings = new InstanceSettings + { + FailingTwitterUserCleanUpThreshold = 300 + }; + #endregion + + #region Mocks + var twitterUserServiceMock = new Mock(MockBehavior.Strict); + twitterUserServiceMock + .Setup(x => x.GetUser(It.IsAny())) + .Returns(new TwitterUser + { + Protected = false + }); + + var twitterUserDalMock = new Mock(MockBehavior.Strict); + var removeTwitterAccountActionMock = new Mock(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 + { + new SyncTwitterUser + { + Id = userId1, + FetchingErrorCount = 100 + } + }; + + var settings = new InstanceSettings + { + FailingTwitterUserCleanUpThreshold = 300 + }; + #endregion + + #region Mocks + var twitterUserServiceMock = new Mock(MockBehavior.Strict); + twitterUserServiceMock + .Setup(x => x.GetUser(It.IsAny())) + .Returns(new TwitterUser + { + Protected = false + }); + + var twitterUserDalMock = new Mock(MockBehavior.Strict); + var removeTwitterAccountActionMock = new Mock(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 + { + 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(MockBehavior.Strict); + twitterUserServiceMock + .Setup(x => x.GetUser(It.Is(y => y == acct1))) + .Returns(new TwitterUser + { + Protected = false + }); + + twitterUserServiceMock + .Setup(x => x.GetUser(It.Is(y => y == acct2))) + .Returns((TwitterUser) null); + + twitterUserServiceMock + .Setup(x => x.PurgeUser(It.Is(y => y == acct2))); + + var twitterUserDalMock = new Mock(MockBehavior.Strict); + twitterUserDalMock + .Setup(x => x.GetTwitterUserAsync(It.Is(y => y == acct2))) + .ReturnsAsync(new SyncTwitterUser + { + Id = userId2, + FetchingErrorCount = 0 + }); + + twitterUserDalMock + .Setup(x => x.UpdateTwitterUserAsync(It.Is(y => y.Id == userId2 && y.FetchingErrorCount == 1))) + .Returns(Task.CompletedTask); + + var removeTwitterAccountActionMock = new Mock(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 + { + 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(MockBehavior.Strict); + twitterUserServiceMock + .Setup(x => x.GetUser(It.Is(y => y == acct1))) + .Returns(new TwitterUser + { + Protected = false + }); + + twitterUserServiceMock + .Setup(x => x.GetUser(It.Is(y => y == acct2))) + .Returns((TwitterUser)null); + + twitterUserServiceMock + .Setup(x => x.PurgeUser(It.Is(y => y == acct2))); + + var twitterUserDalMock = new Mock(MockBehavior.Strict); + twitterUserDalMock + .Setup(x => x.GetTwitterUserAsync(It.Is(y => y == acct2))) + .ReturnsAsync(new SyncTwitterUser + { + Id = userId2, + FetchingErrorCount = 500 + }); + + var removeTwitterAccountActionMock = new Mock(MockBehavior.Strict); + removeTwitterAccountActionMock + .Setup(x => x.ProcessAsync(It.Is(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 + { + 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(MockBehavior.Strict); + twitterUserServiceMock + .Setup(x => x.GetUser(It.Is(y => y == acct1))) + .Returns(new TwitterUser + { + Protected = false + }); + + twitterUserServiceMock + .Setup(x => x.GetUser(It.Is(y => y == acct2))) + .Returns(new TwitterUser + { + Protected = true + }); + + var twitterUserDalMock = new Mock(MockBehavior.Strict); + var removeTwitterAccountActionMock = new Mock(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 + } + } +} \ No newline at end of file diff --git a/src/Tests/BirdsiteLive.Pipeline.Tests/Processors/RetrieveFollowersProcessorTests.cs b/src/Tests/BirdsiteLive.Pipeline.Tests/Processors/RetrieveFollowersProcessorTests.cs index 98a86bf..4679259 100644 --- a/src/Tests/BirdsiteLive.Pipeline.Tests/Processors/RetrieveFollowersProcessorTests.cs +++ b/src/Tests/BirdsiteLive.Pipeline.Tests/Processors/RetrieveFollowersProcessorTests.cs @@ -21,16 +21,16 @@ public async Task ProcessAsync_Test() var userId1 = 1; var userId2 = 2; - var users = new List + var users = new List { - new UserWithTweetsToSync + new UserWithDataToSync { User = new SyncTwitterUser { Id = userId1 } }, - new UserWithTweetsToSync + new UserWithDataToSync { User = new SyncTwitterUser { diff --git a/src/Tests/BirdsiteLive.Pipeline.Tests/Processors/RetrieveTweetsProcessorTests.cs b/src/Tests/BirdsiteLive.Pipeline.Tests/Processors/RetrieveTweetsProcessorTests.cs index 38b750c..17a3aa2 100644 --- a/src/Tests/BirdsiteLive.Pipeline.Tests/Processors/RetrieveTweetsProcessorTests.cs +++ b/src/Tests/BirdsiteLive.Pipeline.Tests/Processors/RetrieveTweetsProcessorTests.cs @@ -4,6 +4,7 @@ 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 @@ public async Task ProcessAsync_UserNotSync_Test() LastTweetPostedId = -1 }; + var user1WtData = new UserWithDataToSync + { + User = user1, + }; + var users = new[] { - user1 + user1WtData }; var tweets = new[] @@ -57,14 +63,12 @@ public async Task ProcessAsync_UserNotSync_Test() It.Is(y => y == user1.Id), It.Is(y => y == tweets.Last().Id), It.Is(y => y == tweets.Last().Id), + It.Is(y => y == 0), It.IsAny() )) .Returns(Task.CompletedTask); var twitterUserServiceMock = new Mock(MockBehavior.Strict); - twitterUserServiceMock - .Setup(x => x.GetUser(It.Is(y => y == user1.Acct))) - .Returns(new TwitterUser {Protected = false}); var logger = new Mock>(MockBehavior.Strict); #endregion @@ -94,9 +98,14 @@ public async Task ProcessAsync_UserSync_Test() LastTweetSynchronizedForAllFollowersId = 46 }; + var user1WtData = new UserWithDataToSync + { + User = user1, + }; + var users = new[] { - user1 + user1WtData }; var tweets = new[] @@ -129,9 +138,6 @@ public async Task ProcessAsync_UserSync_Test() var twitterUserDalMock = new Mock(MockBehavior.Strict); var twitterUserServiceMock = new Mock(MockBehavior.Strict); - twitterUserServiceMock - .Setup(x => x.GetUser(It.Is(y => y == user1.Acct))) - .Returns(new TwitterUser { Protected = false }); var logger = new Mock>(MockBehavior.Strict); #endregion @@ -147,7 +153,7 @@ public async Task ProcessAsync_UserSync_Test() 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 @@ public async Task ProcessAsync_UserPartiallySync_Test() LastTweetSynchronizedForAllFollowersId = 46 }; + var user1WtData = new UserWithDataToSync + { + User = user1, + }; + var users = new[] { - user1 + user1WtData }; var tweets = new[] @@ -199,9 +210,6 @@ public async Task ProcessAsync_UserPartiallySync_Test() var twitterUserDalMock = new Mock(MockBehavior.Strict); var twitterUserServiceMock = new Mock(MockBehavior.Strict); - twitterUserServiceMock - .Setup(x => x.GetUser(It.Is(y => y == user1.Acct))) - .Returns(new TwitterUser { Protected = false }); var logger = new Mock>(MockBehavior.Strict); #endregion @@ -216,7 +224,7 @@ public async Task ProcessAsync_UserPartiallySync_Test() 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 } diff --git a/src/Tests/BirdsiteLive.Pipeline.Tests/Processors/SaveProgressionProcessorTests.cs b/src/Tests/BirdsiteLive.Pipeline.Tests/Processors/SaveProgressionProcessorTests.cs index c95eed6..4587071 100644 --- a/src/Tests/BirdsiteLive.Pipeline.Tests/Processors/SaveProgressionProcessorTests.cs +++ b/src/Tests/BirdsiteLive.Pipeline.Tests/Processors/SaveProgressionProcessorTests.cs @@ -41,7 +41,7 @@ public async Task ProcessAsync_Test() } }; - var usersWithTweets = new UserWithTweetsToSync + var usersWithTweets = new UserWithDataToSync { Tweets = new [] { @@ -65,6 +65,7 @@ public async Task ProcessAsync_Test() It.Is(y => y == user.Id), It.Is(y => y == tweet2.Id), It.Is(y => y == tweet2.Id), + It.Is(y => y == 0), It.IsAny() )) .Returns(Task.CompletedTask); @@ -107,7 +108,7 @@ public async Task ProcessAsync_PartiallySynchronized_Test() } }; - var usersWithTweets = new UserWithTweetsToSync + var usersWithTweets = new UserWithDataToSync { Tweets = new[] { @@ -130,6 +131,7 @@ public async Task ProcessAsync_PartiallySynchronized_Test() It.Is(y => y == user.Id), It.Is(y => y == tweet3.Id), It.Is(y => y == tweet2.Id), + It.Is(y => y == 0), It.IsAny() )) .Returns(Task.CompletedTask); @@ -181,7 +183,7 @@ public async Task ProcessAsync_PartiallySynchronized_MultiUsers_Test() } }; - var usersWithTweets = new UserWithTweetsToSync + var usersWithTweets = new UserWithDataToSync { Tweets = new[] { @@ -205,6 +207,7 @@ public async Task ProcessAsync_PartiallySynchronized_MultiUsers_Test() It.Is(y => y == user.Id), It.Is(y => y == tweet3.Id), It.Is(y => y == tweet2.Id), + It.Is(y => y == 0), It.IsAny() )) .Returns(Task.CompletedTask); diff --git a/src/Tests/BirdsiteLive.Pipeline.Tests/Processors/SendTweetsToFollowersProcessorTests.cs b/src/Tests/BirdsiteLive.Pipeline.Tests/Processors/SendTweetsToFollowersProcessorTests.cs index ad35c3e..53aa12a 100644 --- a/src/Tests/BirdsiteLive.Pipeline.Tests/Processors/SendTweetsToFollowersProcessorTests.cs +++ b/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 @@ public async Task ProcessAsync_SameInstance_SharedInbox_OneTweet_Test() var userId2 = 3; var userAcct = "user"; - var userWithTweets = new UserWithTweetsToSync() + var userWithTweets = new UserWithDataToSync() { Tweets = new [] { @@ -69,15 +71,18 @@ public async Task ProcessAsync_SameInstance_SharedInbox_OneTweet_Test() It.Is(y => y.Length == 2))) .Returns(Task.CompletedTask); + var followersDalMock = new Mock(MockBehavior.Strict); + var loggerMock = new Mock>(); #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 @@ public async Task ProcessAsync_MultiInstances_SharedInbox_OneTweet_Test() var userId2 = 3; var userAcct = "user"; - var userWithTweets = new UserWithTweetsToSync() + var userWithTweets = new UserWithDataToSync() { Tweets = new[] { @@ -139,15 +144,18 @@ public async Task ProcessAsync_MultiInstances_SharedInbox_OneTweet_Test() .Returns(Task.CompletedTask); } + var followersDalMock = new Mock(MockBehavior.Strict); + var loggerMock = new Mock>(); #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 @@ public async Task ProcessAsync_MultiInstances_SharedInbox_OneTweet_Error_Test() var userId2 = 3; var userAcct = "user"; - var userWithTweets = new UserWithTweetsToSync() + var userWithTweets = new UserWithDataToSync() { Tweets = new[] { @@ -214,15 +222,193 @@ public async Task ProcessAsync_MultiInstances_SharedInbox_OneTweet_Error_Test() It.Is(y => y.Length == 1))) .Throws(new Exception()); + var followersDalMock = new Mock(MockBehavior.Strict); + + followersDalMock + .Setup(x => x.UpdateFollowerAsync(It.Is(y => y.Id == userId2 && y.PostingErrorCount == 1))) + .Returns(Task.CompletedTask); + var loggerMock = new Mock>(); #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(MockBehavior.Strict); + + var sendTweetsToSharedInboxTaskMock = new Mock(MockBehavior.Strict); + sendTweetsToSharedInboxTaskMock + .Setup(x => x.ExecuteAsync( + It.Is(y => y.Length == 1), + It.Is(y => y.Acct == userAcct), + It.Is(y => y == host1), + It.Is(y => y.Length == 1))) + .Returns(Task.CompletedTask); + + sendTweetsToSharedInboxTaskMock + .Setup(x => x.ExecuteAsync( + It.Is(y => y.Length == 1), + It.Is(y => y.Acct == userAcct), + It.Is(y => y == host2), + It.Is(y => y.Length == 1))) + .Returns(Task.CompletedTask); + + var followersDalMock = new Mock(MockBehavior.Strict); + + followersDalMock + .Setup(x => x.UpdateFollowerAsync(It.Is(y => y.Id == userId2 && y.PostingErrorCount == 0))) + .Returns(Task.CompletedTask); + + var loggerMock = new Mock>(); + #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(MockBehavior.Strict); + + var sendTweetsToSharedInboxTaskMock = new Mock(MockBehavior.Strict); + sendTweetsToSharedInboxTaskMock + .Setup(x => x.ExecuteAsync( + It.Is(y => y.Length == 1), + It.Is(y => y.Acct == userAcct), + It.Is(y => y == host1), + It.Is(y => y.Length == 1))) + .Returns(Task.CompletedTask); + + sendTweetsToSharedInboxTaskMock + .Setup(x => x.ExecuteAsync( + It.Is(y => y.Length == 1), + It.Is(y => y.Acct == userAcct), + It.Is(y => y == host2), + It.Is(y => y.Length == 1))) + .Throws(new Exception()); + + var followersDalMock = new Mock(MockBehavior.Strict); + + followersDalMock + .Setup(x => x.UpdateFollowerAsync(It.Is(y => y.Id == userId1 && y.PostingErrorCount == 0))) + .Returns(Task.CompletedTask); + + followersDalMock + .Setup(x => x.UpdateFollowerAsync(It.Is(y => y.Id == userId2 && y.PostingErrorCount == 51))) + .Returns(Task.CompletedTask); + + var loggerMock = new Mock>(); + #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 @@ public async Task ProcessAsync_SameInstance_Inbox_OneTweet_Test() var userId2 = 3; var userAcct = "user"; - var userWithTweets = new UserWithTweetsToSync() + var userWithTweets = new UserWithDataToSync() { Tweets = new[] { @@ -282,15 +468,18 @@ public async Task ProcessAsync_SameInstance_Inbox_OneTweet_Test() var sendTweetsToSharedInboxTaskMock = new Mock(MockBehavior.Strict); + var followersDalMock = new Mock(MockBehavior.Strict); + var loggerMock = new Mock>(); #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 @@ public async Task ProcessAsync_MultiInstances_Inbox_OneTweet_Test() var userId2 = 3; var userAcct = "user"; - var userWithTweets = new UserWithTweetsToSync() + var userWithTweets = new UserWithDataToSync() { Tweets = new[] { @@ -351,15 +540,18 @@ public async Task ProcessAsync_MultiInstances_Inbox_OneTweet_Test() var sendTweetsToSharedInboxTaskMock = new Mock(MockBehavior.Strict); + var followersDalMock = new Mock(MockBehavior.Strict); + var loggerMock = new Mock>(); #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 @@ public async Task ProcessAsync_MultiInstances_Inbox_OneTweet_Error_Test() var userId2 = 3; var userAcct = "user"; - var userWithTweets = new UserWithTweetsToSync() + var userWithTweets = new UserWithDataToSync() { Tweets = new[] { @@ -424,15 +616,189 @@ public async Task ProcessAsync_MultiInstances_Inbox_OneTweet_Error_Test() var sendTweetsToSharedInboxTaskMock = new Mock(MockBehavior.Strict); + var followersDalMock = new Mock(MockBehavior.Strict); + + followersDalMock + .Setup(x => x.UpdateFollowerAsync(It.Is(y => y.Id == userId2 && y.PostingErrorCount == 1))) + .Returns(Task.CompletedTask); + + var loggerMock = new Mock>(); + #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_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(MockBehavior.Strict); + sendTweetsToInboxTaskMock + .Setup(x => x.ExecuteAsync( + It.Is(y => y.Length == 1), + It.Is(y => y.Id == userId1), + It.Is(y => y.Acct == userAcct))) + .Returns(Task.CompletedTask); + + sendTweetsToInboxTaskMock + .Setup(x => x.ExecuteAsync( + It.Is(y => y.Length == 1), + It.Is(y => y.Id == userId2), + It.Is(y => y.Acct == userAcct))) + .Returns(Task.CompletedTask); + + var sendTweetsToSharedInboxTaskMock = new Mock(MockBehavior.Strict); + + var followersDalMock = new Mock(MockBehavior.Strict); + + followersDalMock + .Setup(x => x.UpdateFollowerAsync(It.Is(y => y.Id == userId2 && y.PostingErrorCount == 0))) + .Returns(Task.CompletedTask); + + var loggerMock = new Mock>(); + #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(MockBehavior.Strict); + sendTweetsToInboxTaskMock + .Setup(x => x.ExecuteAsync( + It.Is(y => y.Length == 1), + It.Is(y => y.Id == userId1), + It.Is(y => y.Acct == userAcct))) + .Returns(Task.CompletedTask); + + sendTweetsToInboxTaskMock + .Setup(x => x.ExecuteAsync( + It.Is(y => y.Length == 1), + It.Is(y => y.Id == userId2), + It.Is(y => y.Acct == userAcct))) + .Throws(new Exception()); + + var sendTweetsToSharedInboxTaskMock = new Mock(MockBehavior.Strict); + + var followersDalMock = new Mock(MockBehavior.Strict); + + followersDalMock + .Setup(x => x.UpdateFollowerAsync(It.Is(y => y.Id == userId1 && y.PostingErrorCount == 0))) + .Returns(Task.CompletedTask); + + followersDalMock + .Setup(x => x.UpdateFollowerAsync(It.Is(y => y.Id == userId2 && y.PostingErrorCount == 51))) + .Returns(Task.CompletedTask); + var loggerMock = new Mock>(); #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 } } diff --git a/src/Tests/BirdsiteLive.Pipeline.Tests/StatusPublicationPipelineTests.cs b/src/Tests/BirdsiteLive.Pipeline.Tests/StatusPublicationPipelineTests.cs index 2a47b95..81eeb59 100644 --- a/src/Tests/BirdsiteLive.Pipeline.Tests/StatusPublicationPipelineTests.cs +++ b/src/Tests/BirdsiteLive.Pipeline.Tests/StatusPublicationPipelineTests.cs @@ -27,6 +27,7 @@ public async Task ExecuteAsync_Test() It.IsAny())) .Returns(Task.Delay(0)); + var refreshTwitterUserStatusProcessor = new Mock(MockBehavior.Strict); var retrieveTweetsProcessor = new Mock(MockBehavior.Strict); var retrieveFollowersProcessor = new Mock(MockBehavior.Strict); var sendTweetsToFollowersProcessor = new Mock(MockBehavior.Strict); @@ -34,7 +35,7 @@ public async Task ExecuteAsync_Test() var logger = new Mock>(); #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