Background requests for bulk retry, export and bring

SW.Bitween.Api/Controllers/BackgroundRequestsController.cs

@@ -0,0 +1,168 @@
+using System;
+using System.Linq;
+using System.Security.Claims;
+using System.Threading.Tasks;
+using Microsoft.AspNetCore.Authorization;
+using Microsoft.AspNetCore.Http;
+using Microsoft.AspNetCore.Mvc;
+using Microsoft.EntityFrameworkCore;
+using SW.Bitween.Domain;
+using SW.Bitween.Services.Background;
+using SW.PrimitiveTypes;
+
+namespace SW.Bitween.Controllers;
+
+/// <summary>
+/// The signed-in user's background requests (see <see cref="BackgroundRequestMiddleware"/>): which
+/// there are, the answer of one, and cancelling one. Nobody sees anyone else's.
+/// </summary>
+/// <remarks>
+/// A controller rather than CqApi resources, because an answer goes back exactly as the request it
+/// stands in for would have sent it: its status, its content type, its bytes.
+/// </remarks>
+[ApiController]
+[Authorize]
+[Route("api/background")]
+public class BackgroundRequestsController(BitweenDbContext db, BackgroundRequests requests, ICloudFilesService cloudFiles,
+    BackgroundDownloads downloads) : ControllerBase
+{
+    /// <summary>Those still running, and those that finished within the half hour answers are kept.</summary>
+    [HttpGet]
+    public async Task<IActionResult> List()
+    {
+        var owner = Owner;
+        if (owner == null) return Unauthorized();
+        await requests.ForgetExpiredAsync(db);
+
+        var unfinished = await db.Set<BackgroundRequest>()
+            .Where(r => r.Owner == owner &&
+                        (r.Status == BackgroundRequestStatus.Waiting || r.Status == BackgroundRequestStatus.Running))
+            .ToListAsync();
+        await BackgroundRequests.EndStoppedAsync(db, unfinished);
+
+        var now = DateTime.UtcNow;
+        var since = now - BackgroundRequests.KeptFor;
+        var rows = await db.Set<BackgroundRequest>().AsNoTracking()
+            .Where(r => r.Owner == owner && r.StartedOn > now.AddDays(-1) && (r.FinishedOn == null || r.FinishedOn > since))
+            .OrderByDescending(r => r.StartedOn)
+            .Take(20)
+            .Select(r => new
+            {
+                r.Id, r.Label, r.Page, r.Path, r.Status, r.StartedOn, r.FinishedOn, r.ResultStatus, r.Done, r.Total,
+                r.Summary
+            })
+            .ToListAsync();
+
+        return Ok(rows.Select(r => new
+        {
+            r.Id,
+            r.Label,
+            r.Page,
+            r.Path,
+            Status = r.Status.ToString(),
+            r.StartedOn,
+            r.FinishedOn,
+            r.ResultStatus,
+            r.Done,
+            r.Total,
+            r.Summary,
+        }));
+    }
+
+    /// <summary>The answer, as the request would have given it; or why there is none.</summary>
+    [HttpGet("{id}/result")]
+    public async Task<IActionResult> Result(string id)
+    {
+        var owner = Owner;
+        if (owner == null) return Unauthorized();
+        await requests.ForgetExpiredAsync(db);
+
+        var row = await db.Set<BackgroundRequest>().FirstOrDefaultAsync(r => r.Id == id && r.Owner == owner);
+        if (row == null)
+            return Refusal(StatusCodes.Status404NotFound, "BACKGROUND_EXPIRED",
+                "This answer is no longer kept. Open the page again to ask again.");
+
+        if (!row.Finished) await BackgroundRequests.EndStoppedAsync(db, [row]);
+
+        // Work cancelled part way can have an answer too: what it did before it stopped.
+        if (row.Answered)
+        {
+            Response.StatusCode = row.ResultStatus ?? StatusCodes.Status200OK;
+            if (row.ResultContentType != null) Response.ContentType = row.ResultContentType;
+            if (row.ResultDisposition != null) Response.Headers.ContentDisposition = row.ResultDisposition;
+            if (row.ResultKey == null) await Response.Body.WriteAsync(row.Result ?? []);
+            else
+            {
+                await using var kept = await cloudFiles.OpenReadAsync(row.ResultKey);
+                await kept.CopyToAsync(Response.Body);
+            }
+
+            return new EmptyResult();
+        }
+
+        switch (row.Status)
+        {            case BackgroundRequestStatus.Cancelled:
+                return Refusal(StatusCodes.Status410Gone, "BACKGROUND_CANCELLED", "It was cancelled.");
+            case BackgroundRequestStatus.TimedOut:
+                return Refusal(StatusCodes.Status504GatewayTimeout, "BACKGROUND_TIMED_OUT",
+                    $"It went {BackgroundRequests.Limit.TotalMinutes:n0} minutes without getting further and was stopped.");
+            case BackgroundRequestStatus.Stopped:
+                return Refusal(StatusCodes.Status503ServiceUnavailable, "BACKGROUND_STOPPED",
+                    "The server running it stopped before it finished, as a deploy or restart does. Ask again.");
+            default:
+                return Refusal(StatusCodes.Status409Conflict, "BACKGROUND_RUNNING", "It is still running.");
+        }
+    }
+
+    /// <summary>
+    /// A link to the file one of the caller's answered with - an export's zip - for the browser to download
+    /// the ordinary way, straight to disk. It opens for <see cref="BackgroundDownloads.Lasts"/>.
+    /// </summary>
+    [HttpPost("{id}/link")]
+    public async Task<IActionResult> Link(string id)
+    {
+        var owner = Owner;
+        if (owner == null) return Unauthorized();
+        await requests.ForgetExpiredAsync(db);
+
+        var row = await db.Set<BackgroundRequest>().AsNoTracking().FirstOrDefaultAsync(r => r.Id == id && r.Owner == owner);
+        if (row == null)
+            return Refusal(StatusCodes.Status404NotFound, "BACKGROUND_EXPIRED", "This file is no longer kept. Export it again.");
+        if (row.ResultKey == null || row.ResultStatus is not (>= 200 and < 300))
+            return Refusal(StatusCodes.Status409Conflict, "BACKGROUND_NO_FILE", "It has no file to download.");
+        return Ok(new { url = downloads.LinkTo(id) });
+    }
+
+    /// <summary>The file a link from <see cref="Link"/> opens. No sign-in: the link's seal is what lets it in.</summary>
+    [AllowAnonymous]
+    [HttpGet("{id}/file")]
+    public async Task<IActionResult> Download(string id, [FromQuery] string expires, [FromQuery] string seal)
+    {
+        // The same answer for a bad seal as for a missing file, so a link can't be used to probe.
+        if (!downloads.Opens(id, expires, seal)) return NotFound();
+        var row = await db.Set<BackgroundRequest>().AsNoTracking().FirstOrDefaultAsync(r => r.Id == id);
+        if (row?.ResultKey == null) return NotFound();
+
+        if (row.ResultDisposition != null) Response.Headers.ContentDisposition = row.ResultDisposition;
+        return File(await cloudFiles.OpenReadAsync(row.ResultKey), row.ResultContentType ?? "application/octet-stream");
+    }
+
+    /// <summary>Stops one of the caller's, wherever it runs.</summary>
+    [HttpDelete("{id}")]
+    public async Task<IActionResult> Cancel(string id)
+    {
+        var owner = Owner;
+        if (owner == null) return Unauthorized();
+        return await requests.CancelAsync(db, owner, id) ? NoContent() : NotFound();
+    }
+
+    string Owner => User.FindFirst(ClaimTypes.NameIdentifier)?.Value;
+
+    /// <summary>A refusal in the shape the UI reads one: the code, with what to tell the person.</summary>
+    static IActionResult Refusal(int status, string code, string message) => new ContentResult
+    {
+        StatusCode = status,
+        ContentType = "application/json",
+        Content = System.Text.Encoding.UTF8.GetString(BackgroundRequests.Problem(code, message)),
+    };
+}

SW.Bitween.Api/Controllers/XchangeExportController.cs

@@ -44,24 +44,38 @@ public class XchangeExportController(BitweenDbContext dbContext, RequestContext
     : ControllerBase
 {
     /// <summary>
-    /// The most exchanges one zip takes - the bulk retry's ceiling, for the same reason: every file is
-    /// read from storage in sequence, inside the one request.
+    /// The most exchanges one zip takes when asked directly - the bulk retry's ceiling, for the same
+    /// reason: every file is read from storage in sequence, inside the one request.
     /// </summary>
     public const int Limit = 500;
 
+    /// <summary>
+    /// The most one zip takes carried on in the background, as the admin UI asks for it. The zip is built
+    /// on the node's temporary disk and kept in storage, and the browser downloads it from a link, straight
+    /// to disk; <see cref="MaxBytes"/> is what stops a zip of large files first.
+    /// </summary>
+    public const int MaxInBackground = 50_000;
+
+    /// <summary>
+    /// The largest zip built: past it, the exchanges not yet in it are left out, and missing.txt says so.
+    /// Well inside the 2 GB of temporary disk the Helm chart gives a node.
+    /// </summary>
+    public const long MaxBytes = 1L << 30;
+
     public const string MissingList = "missing.txt";
 
     private static readonly Regex Extension = new(@"^\.[A-Za-z0-9]{1,10}$");
 
     /// <summary>
     /// What an export of <paramref name="request"/> would hold, asked before it's downloaded: how many
     /// exchanges it takes, and how many are older than the bucket's rule keeps their files. Refuses what
-    /// the export itself would refuse.
+    /// the export in the background would refuse, past <see cref="MaxInBackground"/>.
     /// </summary>
     [HttpPost("check")]
     public async Task<IActionResult> Check([FromBody] XchangeFilesExport request)
     {
-        var (ids, refusal) = await Resolve(request);
+        // Checked against what the admin UI's export then takes, in the background.
+        var (ids, refusal) = await Resolve(request, MaxInBackground);
         if (refusal != null) return refusal;
 
         var xchanges = await dbContext.Set<Xchange>().AsNoTracking().Where(x => ids.Contains(x.Id)).ToListAsync();
@@ -85,7 +99,8 @@ public async Task<IActionResult> Check([FromBody] XchangeFilesExport request)
     [HttpPost]
     public async Task<IActionResult> Post([FromBody] XchangeFilesExport request)
     {
-        var (ids, refusal) = await Resolve(request);
+        var background = Services.Background.BackgroundRequestScope.Current;
+        var (ids, refusal) = await Resolve(request, background != null ? MaxInBackground : Limit);
         if (refusal != null) return refusal;
 
         var xchanges = await dbContext.Set<Xchange>().AsNoTracking()
@@ -118,8 +133,26 @@ await dbContext.RecordNowAsync(requestContext, OperatorAction.Export,
         await using (var zip = await ZipArchive.CreateAsync(body, ZipArchiveMode.Create, leaveOpen: true,
                          entryNameEncoding: null, aborted))
         {
+            // In the background, as the admin UI asks for it: says how far it has got, and a cancel
+            // ends it between exchanges - with what it took so far, and missing.txt saying so.
+            var exported = 0;
             foreach (var xchange in xchanges)
             {
+                background?.Progress(exported, xchanges.Count, $"{exported:n0} of {xchanges.Count:n0} exchanges");
+                if (background?.CancelRequested.IsCancellationRequested == true)
+                {
+                    missing.Add($"The export was cancelled after {exported} of {xchanges.Count} exchanges; the rest aren't here.");
+                    break;
+                }
+
+                if (body.Written > MaxBytes)
+                {
+                    missing.Add($"The zip reached {MaxBytes / (1024 * 1024 * 1024)} GB after {exported} of {xchanges.Count} exchanges, " +
+                                "the most one export builds; the rest aren't here. Export them separately.");
+                    break;
+                }
+
+                exported++;
                 var mainProperty = mainProperties.GetValueOrDefault(xchange.DocumentId);
                 var mainValue = mainProperty == null ? null : promoted.GetValueOrDefault(xchange.Id)?.Properties?.GetValueOrDefault(mainProperty);
                 var folder = ExchangeArchive.BaseName(mainValue, xchange.Id);
@@ -136,6 +169,8 @@ await Add(zip, xchange, XchangeFileType.Response, $"{folder}/handled",
                     result.NotStoredReason(XchangeFileType.Response));
             }
 
+            background?.Progress(exported, xchanges.Count, $"{exported:n0} of {xchanges.Count:n0} exchanges");
+
             if (missing.Count > 0)
             {
                 await using var list = await zip.CreateEntry(MissingList).OpenAsync(aborted);
@@ -215,8 +250,8 @@ var type when type.StartsWith("text/") => ".txt",
         };
     }
 
-    /// <summary>The exchanges <paramref name="request"/> selects, or why it can't be exported.</summary>
-    private async Task<(List<string> Ids, IActionResult Refusal)> Resolve(XchangeFilesExport request)
+    /// <summary>The exchanges <paramref name="request"/> selects, up to <paramref name="limit"/>, or why it can't be exported.</summary>
+    private async Task<(List<string> Ids, IActionResult Refusal)> Resolve(XchangeFilesExport request, int limit)
     {
         try
         {
@@ -228,13 +263,13 @@ var type when type.StartsWith("text/") => ".txt",
         }
 
         var selection = new XchangeSelection(dbContext);
-        var ids = await selection.Resolve(request, Limit);
+        var ids = await selection.Resolve(request, limit);
         if (ids.Count == 0) return (null, Refuse("NOTHING_SELECTED", "Select at least one exchange to export."));
-        if (ids.Count > Limit)
+        if (ids.Count > limit)
         {
-            var count = string.IsNullOrWhiteSpace(request.Filter) ? ids.Count : await selection.Count(request);
+            var count = string.IsNullOrWhiteSpace(request.Filter) ? ids.Count : await selection.Count(request, limit);
             return (null, Refuse("TOO_MANY",
-                $"{count:n0} exchanges is more than the {Limit} one export can take. Narrow the filter and export the rest after."));
+                $"{count:n0} exchanges is more than the {limit:n0} one export can take. Narrow the filter and export the rest after."));
         }
 
         return (ids, null);
@@ -255,6 +290,9 @@ private sealed class HeldSyncWrites(Stream response) : Stream
     {
         private readonly MemoryStream _held = new();
 
+        /// <summary>How many bytes of zip have been written so far.</summary>
+        public long Written { get; private set; }
+
         public override bool CanRead => false;
         public override bool CanSeek => false;
         public override bool CanWrite => true;
@@ -266,8 +304,17 @@ public override long Position
             set => throw new NotSupportedException();
         }
 
-        public override void Write(byte[] buffer, int offset, int count) => _held.Write(buffer, offset, count);
-        public override void Write(ReadOnlySpan<byte> buffer) => _held.Write(buffer);
+        public override void Write(byte[] buffer, int offset, int count)
+        {
+            _held.Write(buffer, offset, count);
+            Written += count;
+        }
+
+        public override void Write(ReadOnlySpan<byte> buffer)
+        {
+            _held.Write(buffer);
+            Written += buffer.Length;
+        }
 
         /// <summary>Nothing to do until the next asynchronous write or flush.</summary>
         public override void Flush()
@@ -278,6 +325,7 @@ public override async ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, Cancella
         {
             await SendHeld(cancellationToken);
             await response.WriteAsync(buffer, cancellationToken);
+            Written += buffer.Length;
         }
 
         public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) =>

SW.Bitween.Api/Data/BitweenDbContext.cs

@@ -40,12 +40,15 @@ public class BitweenDbContext(DbContextOptions options, RequestContext requestCo
 
         /// <summary>
         /// Every provider's context counts its commands against the exchange stage running, so the
-        /// round trips an exchange costs are measured rather than estimated (see DbCommandTally).
+        /// round trips an exchange costs are measured rather than estimated (see DbCommandTally),
+        /// and lets a background request's commands run past the usual limit (see
+        /// BackgroundRequestScope).
         /// </summary>
         protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder)
         {
             base.OnConfiguring(optionsBuilder);
-            optionsBuilder.AddInterceptors(Services.Metrics.DbCommandCounter.Instance);
+            optionsBuilder.AddInterceptors(Services.Metrics.DbCommandCounter.Instance,
+                Services.Background.BackgroundRequestCommands.Instance);
         }
 
         protected override void OnModelCreating(ModelBuilder modelBuilder)
@@ -395,6 +398,25 @@ protected override void OnModelCreating(ModelBuilder modelBuilder)
                 b.Property(u => u.KeyName).HasMaxLength(500);
             });
 
+            modelBuilder.Entity<BackgroundRequest>(b =>
+            {
+                b.ToTable("BackgroundRequests");
+                b.Property(p => p.Id).HasMaxLength(32).IsUnicode(false);
+                b.Property(p => p.Owner).IsRequired().HasMaxLength(50);
+                b.Property(p => p.Path).IsRequired().HasMaxLength(4000);
+                b.Property(p => p.Label).HasMaxLength(200);
+                b.Property(p => p.Page).HasMaxLength(2000);
+                b.Property(p => p.Node).HasMaxLength(200).IsUnicode(false);
+                b.Property(p => p.ResultContentType).HasMaxLength(200).IsUnicode(false);
+                b.Ignore(p => p.Finished);
+                b.Ignore(p => p.Answered);
+                b.Property(p => p.Summary).HasMaxLength(500);
+                b.Property(p => p.ResultDisposition).HasMaxLength(500);
+                b.Property(p => p.ResultKey).HasMaxLength(500).IsUnicode(false);
+                b.HasIndex(p => new { p.Owner, p.StartedOn });
+                b.HasIndex(p => p.StartedOn);
+            });
+
             modelBuilder.Entity<Domain.Cluster.ClusterNode>(b =>
             {
                 b.ToTable("ClusterNodes");