Major refactor replace MQ with BackgroundJobs

Author: mythz

MyApp.ServiceInterface/AdminServices.cs

@@ -1,13 +1,19 @@
 using Microsoft.AspNetCore.Identity;
 using MyApp.Data;
+using MyApp.ServiceInterface.AiServer;
 using MyApp.ServiceInterface.Renderers;
 using MyApp.ServiceModel;
 using ServiceStack;
+using ServiceStack.Jobs;
 using ServiceStack.OrmLite;
 
 namespace MyApp.ServiceInterface;
 
-public class AdminServices(AppConfig appConfig, ICommandExecutor executor, UserManager<ApplicationUser> userManager,
+public class AdminServices(
+    AppConfig appConfig, 
+    ICommandExecutor executor,
+    IBackgroundJobs jobs,
+    UserManager<ApplicationUser> userManager,
     QuestionsProvider questions)
     : Service
 {
@@ -50,12 +56,10 @@ public async Task<object> Any(Sync request)
     public async Task<object?> Any(GenerateMeta request)
     {
         var regenerateMeta = executor.Command<RegenerateMetaCommand>();
-        await executor.ExecuteAsync(regenerateMeta, new RegenerateMeta
-        {
+        await executor.ExecuteAsync(regenerateMeta, new RegenerateMeta {
             ForPost = request.Id
         });
-        
-        return regenerateMeta.Question;
+        return regenerateMeta.Result;
     }
 
     public async Task<object> Any(AdminResetCommonPassword request)
@@ -122,13 +126,10 @@ public async Task<object> Any(RankAnswer request)
 
         if (answerCreator == null)
             throw HttpError.NotFound($"Answer Creator '{answer.CreatedBy}' not found");
-        
-        MessageProducer.Publish(new AiServerTasks
-        {
-            CreateRankAnswerTask = new CreateRankAnswerTask {
-                AnswerId = answer.RefId!,
-                UserId = answerCreator,
-            } 
+
+        jobs.RunCommand<CreateRankAnswerTaskCommand>(new CreateRankAnswerTask {
+            AnswerId = answer.RefId!,
+            UserId = answerCreator,
         });
 
         return answer;

MyApp.ServiceInterface/AiServer/CreateAnswerCommentTaskCommand.cs

@@ -5,8 +5,19 @@
 
 namespace MyApp.ServiceInterface.AiServer;
 
+public class CreateAnswerCommentTask
+{
+    public string? AiRef { get; set; }
+    public string Model { get; set; }
+    public Post Question { get; set; }
+    public Post Answer { get; set; }
+    public string UserId { get; set; }
+    public string UserName { get; set; }
+    public List<Comment> Comments { get; set; }
+}
+
 [Tag(Tags.AI)]
-public class CreateAnswerCommentTaskCommand(AppConfig appConfig) : IAsyncCommand<CreateAnswerCommentTask>
+public class CreateAnswerCommentTaskCommand(AppConfig appConfig) : AsyncCommand<CreateAnswerCommentTask>
 {
     public const string SystemPrompt = 
         """
@@ -18,7 +29,7 @@ I will provide you with my original question and your initial answer attempt to
         You should use your expertise to provide specific, concise answers to my follow up questions.
         """;
 
-    public async Task ExecuteAsync(CreateAnswerCommentTask request)
+    protected override async Task RunAsync(CreateAnswerCommentTask request, CancellationToken token)
     {
         var question = request.Question;
 

MyApp.ServiceInterface/AiServer/CreateRankAnswerTaskCommand.cs

@@ -4,8 +4,15 @@
 
 namespace MyApp.ServiceInterface.AiServer;
 
+public class CreateRankAnswerTask
+{
+    public string AnswerId { get; set; }
+    public string UserId { get; set; }
+}
+
 [Tag(Tags.AI)]
-public class CreateRankAnswerTaskCommand(AppConfig appConfig, QuestionsProvider questions) : IAsyncCommand<CreateRankAnswerTask>
+public class CreateRankAnswerTaskCommand(AppConfig appConfig, QuestionsProvider questions) 
+    : IAsyncCommand<CreateRankAnswerTask>
 {
     //https://github.com/f/awesome-chatgpt-prompts?tab=readme-ov-file#act-as-a-tech-reviewer
     public const string SystemPrompt = 

MyApp.ServiceInterface/AiServerServices.cs

@@ -4,8 +4,10 @@
 using Microsoft.Extensions.Logging;
 using MyApp.Data;
 using MyApp.ServiceInterface.AiServer;
+using MyApp.ServiceInterface.App;
 using MyApp.ServiceModel;
 using ServiceStack;
+using ServiceStack.Jobs;
 using ServiceStack.OrmLite;
 
 namespace MyApp.ServiceInterface;
@@ -15,7 +17,8 @@ public class AiServerServices(ILogger<AiServerServices> log,
     QuestionsProvider questions, 
     RendererCache rendererCache, 
     WorkerAnswerNotifier answerNotifier,
-    ICommandExecutor executor) : Service
+    ICommandExecutor executor,
+    IBackgroundJobs jobs) : Service
 {
     public async Task<object> Any(CreateAnswersForModels request)
     {
@@ -94,29 +97,23 @@ public async Task Any(CreateAnswerCallback request)
 
         await questions.SaveAnswerAsync(answer);
 
-        MessageProducer.Publish(new DbWrites
+        jobs.RunCommand<SaveGradeResultCommand>(new StatTotals
         {
-            SaveStartingUpVotes = new()
-            {
-                Id = answer.RefId!,
-                PostId = request.PostId,
-                StartingUpVotes = 0,
-                CreatedBy = modelUser.UserName,
-                LastUpdated = DateTime.UtcNow,
-            }
+            Id = answer.RefId!,
+            PostId = request.PostId,
+            StartingUpVotes = 0,
+            CreatedBy = modelUser.UserName,
+            LastUpdated = DateTime.UtcNow,
         });
 
-        await Db.NotifyQuestionAuthorIfRequiredAsync(MessageProducer, answer);
-        
-        MessageProducer.Publish(new AiServerTasks
-        {
-            CreateRankAnswerTask = new CreateRankAnswerTask {
-                AnswerId = answer.RefId!,
-                UserId = request.UserId,
-            } 
+        await Db.NotifyQuestionAuthorIfRequiredAsync(jobs, answer);
+
+        jobs.RunCommand<CreateRankAnswerTaskCommand>(new CreateRankAnswerTask {
+            AnswerId = answer.RefId!,
+            UserId = request.UserId,
         });
-        
-        MessageProducer.Publish(new SearchTasks {
+
+        jobs.RunCommand<SearchTasksCommand>(new SearchTasks {
             AddAnswerToIndex = answer.RefId
         });
     }
@@ -173,11 +170,8 @@ public async Task Any(RankAnswerCallback request)
             meta.StatTotals.Add(statTotals);
 
             await questions.SaveMetaAsync(request.PostId, meta);
-            
-            MessageProducer.Publish(new DbWrites
-            {
-                SaveStartingUpVotes = statTotals 
-            });
+
+            jobs.RunCommand<SaveGradeResultCommand>(statTotals);
         }
         catch (Exception e)
         {
@@ -223,14 +217,10 @@ public async Task Any(AnswerCommentCallback request)
         };
         comments.Add(newComment);
 
-        MessageProducer.Publish(new DbWrites
-        {
-            NewComment = new()
-            {
-                RefId = request.AnswerId,
-                Comment = newComment,
-                LastUpdated = DateTime.UtcNow,
-            },
+        jobs.RunCommand<NewCommentCommand>(new NewComment {
+            RefId = request.AnswerId,
+            Comment = newComment,
+            LastUpdated = DateTime.UtcNow,
         });
 
         await questions.SaveMetaAsync(postId, meta);

MyApp.ServiceInterface/AnalyticsTasksCommand.cs

@@ -0,0 +1,34 @@
+using MyApp.Data;
+using MyApp.ServiceModel;
+using ServiceStack;
+using ServiceStack.IO;
+using ServiceStack.OrmLite;
+
+namespace MyApp.ServiceInterface;
+
+[Worker(Databases.Analytics)]
+public class AnalyticsTasksCommand(R2VirtualFiles r2, QuestionsProvider questions) : AsyncCommand<AnalyticsTasks>
+{
+    protected override async Task RunAsync(AnalyticsTasks request, CancellationToken token)
+    {
+        if (request.CreatePostStat == null && request.CreateSearchStat == null && request.DeletePost == null)
+            return;
+
+        using var analyticsDb = HostContext.AppHost.GetDbConnection(Databases.Analytics);
+        
+        if (request.CreatePostStat != null)// && !Stats.IsAdminOrModerator(request.RecordPostView.UserName))
+        {
+            await analyticsDb.InsertAsync(request.CreatePostStat, token: token);
+        }
+
+        if (request.CreateSearchStat != null)// && !Stats.IsAdminOrModerator(request.RecordSearchView.UserName))
+        {
+            await analyticsDb.InsertAsync(request.CreateSearchStat, token: token);
+        }
+
+        if (request.DeletePost != null)
+        {
+            await analyticsDb.DeleteAsync<PostStat>(x => x.PostId == request.DeletePost, token: token);
+        }
+    }
+}

MyApp.ServiceInterface/App/AnswerAddedToPostCommand.cs

@@ -7,12 +7,13 @@
 namespace MyApp.ServiceInterface.App;
 
 [Tag(Tags.Answers)]
-public class AnswerAddedToPostCommand(IDbConnection db) : IAsyncCommand<AnswerAddedToPost>
+[Worker(Databases.App)]
+public class AnswerAddedToPostCommand(IDbConnection db) : AsyncCommand<AnswerAddedToPost>
 {
-    public async Task ExecuteAsync(AnswerAddedToPost request)
+    protected override async Task RunAsync(AnswerAddedToPost request, CancellationToken token)
     {
         await db.UpdateAddAsync(() => new Post {
             AnswerCount = 1,
-        }, x => x.Id == request.Id);
+        }, x => x.Id == request.Id, token:token);
     }
 }

MyApp.ServiceInterface/App/CreateAnswerCommand.cs

@@ -7,6 +7,7 @@
 namespace MyApp.ServiceInterface.App;
 
 [Tag(Tags.Answers)]
+[Worker(Databases.App)]
 public class CreateAnswerCommand(AppConfig appConfig, IDbConnection db) : IAsyncCommand<Post>
 {
     public async Task ExecuteAsync(Post answer)

MyApp.ServiceInterface/App/CreateFlagCommand.cs

@@ -6,10 +6,11 @@
 namespace MyApp.ServiceInterface.App;
 
 [Tag(Tags.Database)]
-public class CreateFlagCommand(IDbConnection db) : IAsyncCommand<Flag>
+[Worker(Databases.App)]
+public class CreateFlagCommand(IDbConnection db) : AsyncCommand<Flag>
 {
-    public async Task ExecuteAsync(Flag request)
+    protected override async Task RunAsync(Flag request, CancellationToken token)
     {
-        await db.InsertAsync(request);
+        await db.InsertAsync(request, token: token);
     }
 }

MyApp.ServiceInterface/App/CreateNotificationCommand.cs

@@ -7,11 +7,12 @@
 namespace MyApp.ServiceInterface.App;
 
 [Tag(Tags.Notifications)]
-public class CreateNotificationCommand(AppConfig appConfig, IDbConnection db) : IAsyncCommand<Notification>
+[Worker(Databases.App)]
+public class CreateNotificationCommand(AppConfig appConfig, IDbConnection db) : AsyncCommand<Notification>
 {
-    public async Task ExecuteAsync(Notification request)
+    protected override async Task RunAsync(Notification request, CancellationToken token)
     {
-        await db.InsertAsync(request);
+        await db.InsertAsync(request, token: token);
         appConfig.IncrUnreadNotificationsFor(request.UserName);
     }
-}
+}

MyApp.ServiceInterface/App/CreatePostCommand.cs

@@ -7,21 +7,22 @@
 
 namespace MyApp.ServiceInterface.App;
 
+[Worker(Databases.App)]
 [Tag(Tags.Questions)]
-public class CreatePostCommand(ILogger<CreatePostCommand> log, AppConfig appConfig, IDbConnection db) : IAsyncCommand<Post>
+public class CreatePostCommand(ILogger<CreatePostCommand> log, AppConfig appConfig, IDbConnection db) : AsyncCommand<Post>
 {
-    public async Task ExecuteAsync(Post post)
+    protected override async Task RunAsync(Post post, CancellationToken token)
     {
         var body = post.Body;
         post.Body = null;
 
         if (post.Id > 0)
         {
-            await db.InsertAsync(post);
+            await db.InsertAsync(post, token: token);
         }
         else
         {
-            post.Id = (int)await db.InsertAsync(post, selectIdentity: true);
+            post.Id = (int)await db.InsertAsync(post, selectIdentity: true, token: token);
         }
 
         var createdBy = post.CreatedBy;
@@ -40,7 +41,7 @@ await db.InsertAsync(new StatTotals
                 DownVotes = 0,
                 StartingUpVotes = 0,
                 CreatedBy = post.CreatedBy,
-            });
+            }, token: token);
         }
         catch (Exception e)
         {
@@ -50,7 +51,7 @@ await db.InsertAsync(new StatTotals
             {
                 PostId = post.Id,
                 CreatedBy = post.CreatedBy,
-            }, x => x.Id == $"{post.Id}");
+            }, x => x.Id == $"{post.Id}", token: token);
         }
 
         if (!string.IsNullOrEmpty(body))
@@ -61,7 +62,7 @@ await db.InsertAsync(new StatTotals
             if (userNameMentions.Count > 0)
             {
                 var existingUsers = await db.SelectAsync(db.From<ApplicationUser>()
-                    .Where(x => userNameMentions.Contains(x.UserName!)));
+                    .Where(x => userNameMentions.Contains(x.UserName!)), token: token);
 
                 foreach (var existingUser in existingUsers)
                 {
@@ -80,7 +81,7 @@ await db.InsertAsync(new Notification
                             CreatedDate = post.CreationDate,
                             Summary = cleanBody.GenerateNotificationSummary(startPos),
                             RefUserName = createdBy,
-                        });
+                        }, token: token);
                         appConfig.IncrUnreadNotificationsFor(existingUser.UserName!);
                     }
                 }
@@ -97,7 +98,7 @@ await db.InsertAsync(new Achievement
                 PostId = post.Id,
                 Score = 1,
                 CreatedDate = DateTime.UtcNow,
-            });
+            }, token: token);
             appConfig.IncrUnreadAchievementsFor(post.CreatedBy!);
 
             // Setup auto-watch for new questions (Sending Emails for new Answers)
@@ -108,7 +109,7 @@ await db.InsertAsync(new WatchPost
                 CreatedDate = post.CreationDate,
                 // Email new answers 1hr after asking question
                 AfterDate = DateTime.UtcNow.Add(TimeSpan.FromHours(1)),
-            });
+            }, token: token);
         }
     }
 }