raw iteration of sending notes in pipeline

src/BirdsiteLive.Domain/ActivityPubService.cs

@@ -4,6 +4,7 @@ using System.Net.Http;
 using System.Text;
 using System.Threading.Tasks;
 using BirdsiteLive.ActivityPub;
+using BirdsiteLive.Common.Settings;
 using Newtonsoft.Json;
 using Org.BouncyCastle.Bcpg;
 
@@ -13,16 +14,20 @@ namespace BirdsiteLive.Domain
     {
         Task<Actor> GetUser(string objectId);
         Task<HttpStatusCode> PostDataAsync<T>(T data, string targetHost, string actorUrl, string inbox = null);
+        Task<HttpStatusCode> PostNewNoteActivity(Note note, string username, string noteId, string targetHost,
+            string targetInbox);
     }
 
     public class ActivityPubService : IActivityPubService
     {
+        private readonly InstanceSettings _instanceSettings;
         private readonly ICryptoService _cryptoService;
 
         #region Ctor
-        public ActivityPubService(ICryptoService cryptoService)
+        public ActivityPubService(ICryptoService cryptoService, InstanceSettings instanceSettings)
         {
             _cryptoService = cryptoService;
+            _instanceSettings = instanceSettings;
         }
         #endregion
 
@@ -37,6 +42,55 @@ namespace BirdsiteLive.Domain
             }
         }
 
+        public async Task<HttpStatusCode> PostNewNoteActivity(Note note, string username, string noteId, string targetHost, string targetInbox)
+        {
+            //var username = "gra";
+            var actor = $"https://{_instanceSettings.Domain}/users/{username}";
+            //var targetHost = "mastodon.technology";
+            //var target = $"{targetHost}/users/testtest";
+            //var inbox = $"/users/testtest/inbox";
+
+            //var noteGuid = Guid.NewGuid();
+            var noteUri = $"https://{_instanceSettings.Domain}/users/{username}/statuses/{noteId}";
+            
+            //var noteUrl = $"https://{_instanceSettings.Domain}/@{username}/{noteId}";
+            //var to = $"{actor}/followers";
+            //var apPublic = "https://www.w3.org/ns/activitystreams#Public";
+
+            var now = DateTime.UtcNow;
+            var nowString = now.ToString("s") + "Z";
+            
+            var noteActivity = new ActivityCreateNote()
+            {
+                context = "https://www.w3.org/ns/activitystreams",
+                id = $"{noteUri}/activity",
+                type = "Create",
+                actor = actor,
+                published = nowString,
+
+                to = note.to,
+                cc = note.cc,
+                apObject = note
+                //apObject = new Note()
+                //{
+                //    id = noteUri,
+                //    summary = null,
+                //    inReplyTo = null,
+                //    published = nowString,
+                //    url = noteUrl,
+                //    attributedTo = actor,
+                //    to = new[] { to },
+                //    //cc = new [] { apPublic },
+                //    sensitive = false,
+                //    content = "<p>Woooot</p>",
+                //    attachment = new string[0],
+                //    tag = new string[0]
+                //}
+            };
+
+            return await PostDataAsync(noteActivity, targetHost, actor, targetInbox);
+        }
+
         public async Task<HttpStatusCode> PostDataAsync<T>(T data, string targetHost, string actorUrl, string inbox = null)
         {
             var usedInbox = $"/inbox";

src/BirdsiteLive.Pipeline/BirdsiteLive.Pipeline.csproj

@@ -6,11 +6,12 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Microsoft.Bcl.AsyncInterfaces" Version="1.0.0" />
+    <PackageReference Include="Microsoft.Bcl.AsyncInterfaces" Version="1.1.1" />
     <PackageReference Include="System.Threading.Tasks.Dataflow" Version="4.11.1" />
   </ItemGroup>
 
   <ItemGroup>
+    <ProjectReference Include="..\BirdsiteLive.Domain\BirdsiteLive.Domain.csproj" />
     <ProjectReference Include="..\BirdsiteLive.Twitter\BirdsiteLive.Twitter.csproj" />
     <ProjectReference Include="..\DataAccessLayers\BirdsiteLive.DAL\BirdsiteLive.DAL.csproj" />
   </ItemGroup>

src/BirdsiteLive.Pipeline/Processors/SendTweetsToFollowersProcessor.cs

@@ -1,15 +1,60 @@
-using System.Threading;
+using System.Linq;
+using System.Net;
+using System.Threading;
 using System.Threading.Tasks;
+using BirdsiteLive.DAL.Contracts;
+using BirdsiteLive.Domain;
 using BirdsiteLive.Pipeline.Contracts;
 using BirdsiteLive.Pipeline.Models;
+using BirdsiteLive.Twitter;
 
 namespace BirdsiteLive.Pipeline.Processors
 {
     public class SendTweetsToFollowersProcessor : ISendTweetsToFollowersProcessor
     {
-        public Task ProcessAsync(UserWithTweetsToSync userWithTweetsToSync, CancellationToken ct)
+        private readonly IActivityPubService _activityPubService;
+        private readonly IUserService _userService;
+        private readonly IFollowersDal _followersDal;
+        private readonly ITwitterUserDal _twitterUserDal;
+
+        #region Ctor
+        public SendTweetsToFollowersProcessor(IActivityPubService activityPubService, IUserService userService, IFollowersDal followersDal, ITwitterUserDal twitterUserDal)
         {
-            throw new System.NotImplementedException();
+            _activityPubService = activityPubService;
+            _userService = userService;
+            _followersDal = followersDal;
+            _twitterUserDal = twitterUserDal;
+        }
+        #endregion
+
+        public async Task ProcessAsync(UserWithTweetsToSync userWithTweetsToSync, CancellationToken ct)
+        {
+            var user = userWithTweetsToSync.User;
+            var userId = user.Id;
+
+            foreach (var follower in userWithTweetsToSync.Followers)
+            {
+                var fromStatusId = follower.FollowingsSyncStatus[userId];
+                var tweetsToSend = userWithTweetsToSync.Tweets.Where(x => x.Id > fromStatusId).OrderBy(x => x.Id).ToList();
+
+                var syncStatus = fromStatusId;
+                foreach (var tweet in tweetsToSend)
+                {
+                    var note = _userService.GetStatus(user.Acct, tweet);
+                    var result = await _activityPubService.PostNewNoteActivity(note, user.Acct, tweet.Id.ToString(), follower.Host, follower.InboxUrl);
+                    if (result == HttpStatusCode.Accepted)
+                        syncStatus = tweet.Id;
+                    else
+                        break;
+                }
+
+                follower.FollowingsSyncStatus[userId] = syncStatus;
+                await _followersDal.UpdateFollowerAsync(follower);
+            }
+
+            var lastPostedTweet = userWithTweetsToSync.Tweets.Select(x => x.Id).Max();
+            var minimumSync = userWithTweetsToSync.Followers.Select(x => x.FollowingsSyncStatus[userId]).Min();
+            await _twitterUserDal.UpdateTwitterUserAsync(userId, lastPostedTweet, minimumSync);
         }
     }
 }