Merge pull request #317 from peppy/reprocess-score-command

Add command to reprocess a single score

osu.Server.Queues.ScoreStatisticsProcessor/Commands/Queue/ProcessScoreCommand.cs

@@ -0,0 +1,81 @@
+// Copyright (c) ppy Pty Ltd <[email protected]>. Licensed under the MIT Licence.
+// See the LICENCE file in the repository root for full licence text.
+
+using System;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using Dapper;
+using McMaster.Extensions.CommandLineUtils;
+using osu.Server.QueueProcessor;
+using osu.Server.Queues.ScoreStatisticsProcessor.Models;
+
+namespace osu.Server.Queues.ScoreStatisticsProcessor.Commands.Queue
+{
+    [Command("process-single", Description = "Process a specific single score.")]
+    public class ProcessScoreCommand
+    {
+        /// <summary>
+        /// A comma-separated list of processors to disable.
+        /// </summary>
+        [Option("--disable", Description = "A comma-separated list of processors to disable.")]
+        public string DisabledProcessors { get; set; } = string.Empty;
+
+        [Option("--force", Description = "When set, reprocess the score even if it was already processed up to the current version.")]
+        public bool Force { get; set; } = false;
+
+        [Argument(0, Description = "The ID of the score to process")]
+        public long ScoreId { get; set; }
+
+        public async Task<int> OnExecuteAsync(CancellationToken cancellationToken)
+        {
+            string? disabledProcessorsEnv = Environment.GetEnvironmentVariable("DISABLED_PROCESSORS");
+
+            if (!string.IsNullOrEmpty(disabledProcessorsEnv))
+            {
+                if (!string.IsNullOrEmpty(DisabledProcessors))
+                    throw new ArgumentException("Attempted to specify disabled processors via parameter and environment at the same time");
+
+                DisabledProcessors = disabledProcessorsEnv;
+            }
+
+            ScoreStatisticsQueueProcessor queue = new ScoreStatisticsQueueProcessor(DisabledProcessors.Split(',', StringSplitOptions.TrimEntries | StringSplitOptions.RemoveEmptyEntries));
+
+            using (var db = await DatabaseAccess.GetConnectionAsync(cancellationToken))
+            {
+                var score = await db.QuerySingleOrDefaultAsync<SoloScore?>("SELECT * FROM scores WHERE id = @score_id", new { score_id = ScoreId });
+                var history = await db.QuerySingleOrDefaultAsync<ProcessHistory?>("SELECT * FROM score_process_history WHERE score_id = @score_id", new { score_id = ScoreId });
+
+                if (score == null)
+                {
+                    Console.WriteLine($"Score ID {ScoreId} doesn't exist.");
+                    return 1;
+                }
+
+                Console.WriteLine($"Processing score {score}");
+
+                Console.WriteLine(history == null
+                    ? "- Score has not yet been processed"
+                    : $"- Attaching process history with version {history.processed_version}");
+
+                var scoreItem = new ScoreItem(score, history);
+
+                queue.ProcessScore(scoreItem, Force);
+
+                Console.WriteLine();
+
+                if (scoreItem.Tags?.Any() == true)
+                {
+                    Console.WriteLine("Processing completed with tags:");
+                    Console.WriteLine(string.Join(", ", scoreItem.Tags));
+                }
+                else
+                {
+                    Console.WriteLine("Processing completed but no tags were appended.");
+                }
+            }
+
+            return 0;
+        }
+    }
+}

osu.Server.Queues.ScoreStatisticsProcessor/Commands/QueueCommands.cs

@@ -11,6 +11,7 @@ namespace osu.Server.Queues.ScoreStatisticsProcessor.Commands
     [Command(Name = "queue", Description = "Perform various operations on the processing queue.")]
     [Subcommand(typeof(PumpTestDataCommand))]
     [Subcommand(typeof(PumpAllScoresCommand))]
+    [Subcommand(typeof(ProcessScoreCommand))]
     [Subcommand(typeof(ClearQueueCommand))]
     [Subcommand(typeof(WatchQueueCommand))]
     [Subcommand(typeof(ImportHighScoresCommand))]

osu.Server.Queues.ScoreStatisticsProcessor/ScoreStatisticsQueueProcessor.cs

@@ -154,7 +154,16 @@ private List<IProcessor> createProcessors(string[]? disabledProcessors)
             return instances.OrderBy(processor => processor.Order).ToList();
         }
 
-        protected override void ProcessResult(ScoreItem item)
+        /// <summary>
+        /// Process the provided score item.
+        /// </summary>
+        /// <param name="item">The score to process.</param>
+        /// <param name="force">Whether to process regardless of whether the attached process history implies it is already processed up-to-date.</param>
+        public void ProcessScore(ScoreItem item, bool force) => processScore(item, force);
+
+        protected override void ProcessResult(ScoreItem item) => processScore(item, false);
+
+        private void processScore(ScoreItem item, bool force = false)
         {
             var stopwatch = new Stopwatch();
             var tags = new List<string>();
@@ -175,7 +184,7 @@ protected override void ProcessResult(ScoreItem item)
                 if (item.Score.legacy_score_id != null)
                     tags.Add("type:legacy");
 
-                if (item.ProcessHistory?.processed_version == VERSION)
+                if (item.ProcessHistory?.processed_version == VERSION && !force)
                 {
                     tags.Add("type:skipped");
                     return;