Aggregation as a first-class integration, and table filters

SW.Bitween.Api/Resources/Subscriptions/Search.cs

@@ -56,6 +56,11 @@ join document in _dbContext.Set<Document>() on subscriber.DocumentId equals docu
                     AggregationForId = subscriber.AggregationForId,
                     Temporary = subscriber.Temporary,
                     ReceiveOn = subscriber.ReceiveOn,
+                    // The next-fire time of the other scheduled type. Left out, an aggregation
+                    // row had no next run to show at all — ReceiveOn is only ever set for
+                    // Receiving, and the list page reads one field for both.
+                    AggregateOn = subscriber.AggregateOn,
+                    AggregationTarget = subscriber.AggregationTarget,
                     PausedOn = subscriber.PausedOn,
                     IsRunning = subscriber.IsRunning,
                     ConsecutiveFailures = subscriber.ConsecutiveFailures,

SW.Bitween.Api/Resources/Subscriptions/SubscriptionConfigurationApplier.cs

@@ -37,6 +37,9 @@ public static async Task Apply(BitweenDbContext dbContext, Subscription entity,
         entity.WorkGroupId = model.WorkGroupId;
         entity.ResponseSubscriptionId = model.ResponseSubscriptionId;
         entity.ResponseMessageTypeName = model.ResponseMessageTypeName;
+        // Meaningless for every other type, where it stays at its default — but harmless
+        // there, and applying it unconditionally is what stops create and update disagreeing.
+        entity.AggregationTarget = model.AggregationTarget;
 
         // Only when the caller said something about them. A Receiving subscription with no
         // schedules is what SetSchedules throws on, and a create that mentions no schedule at

SW.Bitween.Api/Services/AggregationJob.cs

@@ -2,6 +2,7 @@
 using Microsoft.Extensions.Logging;
 using Newtonsoft.Json;
 using SW.Bitween.Domain;
+using SW.Bitween.Model;
 using SW.Scheduler;
 using System;
 using System.Linq;
@@ -25,6 +26,15 @@ public async Task Execute(AggregationJobParams jobParams)
 
         if (aggSub == null) return;
 
+        var startedOn = DateTime.UtcNow;
+        // Decided as the run goes and written once at the end. Recording inside both the try
+        // and the catch wrote two rows for a single run whenever something threw after the
+        // roll-up was already built — SetSchedules does exactly that on an aggregation that
+        // somehow has no schedule.
+        var outcome = ReceiveOutcome.NoNewData;
+        string errorMessage = null;
+        string[] createdExchangeIds = [];
+
         try
         {
             var xchangeQuery =
@@ -46,17 +56,47 @@ from agg in xa.DefaultIfEmpty()
                 var aggXchange = await xchangeService.CreateXchange(aggSub, xchangeAggregationFile);
                 dbContext.Add(aggXchange);
                 targetXchangeList.ForEach(id => dbContext.Add(new XchangeAggregation(id, aggXchange.Id)));
+                outcome = ReceiveOutcome.Received;
+                createdExchangeIds = [aggXchange.Id];
             }
+            // Otherwise the outcome stays NoNewData: nothing outstanding is an ordinary, quiet
+            // result, the same thing a receiver reports when the folder it polls is empty.
 
             aggSub.SetSchedules();
             aggSub.SetHealth();
         }
         catch (Exception ex)
         {
             aggSub.SetHealth(ex.ToString());
+            errorMessage = ex.ToString();
+            // A run that already built its roll-up did the thing it exists to do; a throw after
+            // that is bookkeeping, and calling the run failed would misreport delivered work.
+            // The error is kept on the attempt either way, and on the integration's health.
+            if (createdExchangeIds.Length == 0) outcome = ReceiveOutcome.Failed;
             logger.LogError(ex, "Error processing aggregation for subscription {SubscriptionId}", jobParams.SubscriptionId);
         }
 
+        RecordAttempt(aggSub.Id, startedOn, outcome, errorMessage, createdExchangeIds);
+
         await dbContext.SaveChangesAsync();
     }
+
+    /// <summary>
+    /// Writes the run into the same history a receiver's runs go into, which is what lets one
+    /// table on the integration page show a run beside the exchange it produced. The entity is
+    /// named for the job that came first; nothing on it is specific to receiving.
+    /// </summary>
+    private void RecordAttempt(int subscriptionId, DateTime startedOn, ReceiveOutcome outcome,
+        string errorMessage, string[] exchangeIds)
+    {
+        dbContext.Add(new ReceiveAttempt
+        {
+            SubscriptionId = subscriptionId,
+            StartedOn = startedOn,
+            FinishedOn = DateTime.UtcNow,
+            Outcome = outcome,
+            ErrorMessage = errorMessage,
+            ExchangeIds = exchangeIds,
+        });
+    }
 }

SW.Bitween.IntegrationTests/Tests/AggregationTests.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
+using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.EntityFrameworkCore;
 using Microsoft.Extensions.DependencyInjection;
@@ -157,4 +158,241 @@ public async Task Aggregation_job_does_nothing_for_inactive_subscription()
         var count = await db.Set<Xchange>().CountAsync(x => x.SubscriptionId == aggSub.Id);
         Assert.Equal(0, count);
     }
+
+    // ─── Run history ────────────────────────────────────────────────────────
+    //
+    // Every run is written into the same history a receiver's runs go into, so the integration
+    // page can show one table of runs with the exchange each produced instead of two tables
+    // that cannot be joined. Before this, an aggregation recorded nothing at all.
+
+    [Fact]
+    public async Task A_run_that_rolls_something_up_records_the_exchange_it_made()
+    {
+        await using var scope = _fixture.CreateScope();
+        var db = scope.ServiceProvider.GetRequiredService<BitweenDbContext>();
+        var job = scope.ServiceProvider.GetRequiredService<AggregationJob>();
+        var cache = _fixture.App.Services.GetRequiredService<IInfolinkCache>();
+
+        var sourceDoc = new Document(null, "Agg Attempt Doc", DocumentFormat.Json);
+        db.Set<Document>().Add(sourceDoc);
+        await db.SaveChangesAsync();
+        var sourceSub = new Subscription("Agg Attempt Source", sourceDoc.Id);
+        sourceSub.Inactive = false;
+        db.Set<Subscription>().Add(sourceSub);
+        await db.SaveChangesAsync();
+
+        var xchange = new Xchange(sourceSub, new XchangeFile("{\"n\":1}"));
+        db.Set<Xchange>().Add(xchange);
+        await db.SaveChangesAsync();
+        db.Set<XchangeResult>().Add(new XchangeResult(xchange.Id, null, null));
+        await db.SaveChangesAsync();
+
+        var aggSub = new Subscription("Agg Attempt", sourceSub.Id, Partner.SystemId);
+        aggSub.Inactive = false;
+        // A real aggregation always has one — SetSchedules throws without it, and the run would
+        // then report a failure it did not have.
+        aggSub.SetSchedules([new Schedule(Recurrence.Daily, TimeSpan.FromHours(2))]);
+        db.Set<Subscription>().Add(aggSub);
+        await db.SaveChangesAsync();
+
+        cache.Revoke();
+
+        await job.Execute(new AggregationJobParams(aggSub.Id, null));
+
+        var attempt = await db.Set<ReceiveAttempt>().SingleAsync(a => a.SubscriptionId == aggSub.Id);
+        Assert.Equal(ReceiveOutcome.Received, attempt.Outcome);
+        Assert.Null(attempt.ErrorMessage);
+
+        var rollUp = await db.Set<Xchange>().SingleAsync(x => x.SubscriptionId == aggSub.Id);
+        Assert.Equal([rollUp.Id], attempt.ExchangeIds);
+    }
+
+    [Fact]
+    public async Task A_run_with_nothing_outstanding_records_no_new_data_rather_than_nothing()
+    {
+        await using var scope = _fixture.CreateScope();
+        var db = scope.ServiceProvider.GetRequiredService<BitweenDbContext>();
+        var job = scope.ServiceProvider.GetRequiredService<AggregationJob>();
+        var cache = _fixture.App.Services.GetRequiredService<IInfolinkCache>();
+
+        var sourceDoc = new Document(null, "Agg Empty Attempt Doc", DocumentFormat.Json);
+        db.Set<Document>().Add(sourceDoc);
+        await db.SaveChangesAsync();
+        // A source with no exchanges at all — the run has nothing to collect.
+        var sourceSub = new Subscription("Agg Empty Attempt Source", sourceDoc.Id);
+        sourceSub.Inactive = false;
+        db.Set<Subscription>().Add(sourceSub);
+        await db.SaveChangesAsync();
+
+        var aggSub = new Subscription("Agg Empty Attempt", sourceSub.Id, Partner.SystemId);
+        aggSub.Inactive = false;
+        // A real aggregation always has one — SetSchedules throws without it, and the run would
+        // then report a failure it did not have.
+        aggSub.SetSchedules([new Schedule(Recurrence.Daily, TimeSpan.FromHours(2))]);
+        db.Set<Subscription>().Add(aggSub);
+        await db.SaveChangesAsync();
+
+        cache.Revoke();
+
+        await job.Execute(new AggregationJobParams(aggSub.Id, null));
+
+        var attempt = await db.Set<ReceiveAttempt>().SingleAsync(a => a.SubscriptionId == aggSub.Id);
+        Assert.Equal(ReceiveOutcome.NoNewData, attempt.Outcome);
+        Assert.Empty(attempt.ExchangeIds);
+        // The quiet outcome still creates no exchange — that part has not changed.
+        Assert.Equal(0, await db.Set<Xchange>().CountAsync(x => x.SubscriptionId == aggSub.Id));
+    }
+
+    [Fact]
+    public async Task An_inactive_aggregation_records_no_run_at_all()
+    {
+        await using var scope = _fixture.CreateScope();
+        var db = scope.ServiceProvider.GetRequiredService<BitweenDbContext>();
+        var job = scope.ServiceProvider.GetRequiredService<AggregationJob>();
+
+        var sourceDoc = new Document(null, "Agg Skipped Attempt Doc", DocumentFormat.Json);
+        db.Set<Document>().Add(sourceDoc);
+        await db.SaveChangesAsync();
+        var sourceSub = new Subscription("Agg Skipped Attempt Source", sourceDoc.Id);
+        db.Set<Subscription>().Add(sourceSub);
+        await db.SaveChangesAsync();
+
+        // Inactive by default — the job returns before it does anything, and a run that never
+        // happened must not appear in the history as a quiet success.
+        var aggSub = new Subscription("Agg Skipped Attempt", sourceSub.Id, Partner.SystemId);
+        db.Set<Subscription>().Add(aggSub);
+        await db.SaveChangesAsync();
+
+        await job.Execute(new AggregationJobParams(aggSub.Id, null));
+
+        Assert.Empty(await db.Set<ReceiveAttempt>().Where(a => a.SubscriptionId == aggSub.Id).ToListAsync());
+    }
+
+    // ─── Configuration ──────────────────────────────────────────────────────
+    //
+    // Which file a roll-up links to is the one aggregation setting no UI has ever offered.
+    // It reached the entity only through the generic property copy in the update handler, so
+    // create could not set it at all and every aggregation started on Input whatever the
+    // caller asked for. These pin the field to the shared applier both handlers run.
+
+    private static int _seq;
+    private static string Unique(string prefix) => $"{prefix}-{Interlocked.Increment(ref _seq)}";
+
+    /// <summary>An integration to roll up, and a partner to attribute the roll-up to.</summary>
+    private async Task<(int sourceId, int partnerId)> Groundwork()
+    {
+        await using var scope = _fixture.CreateScope();
+        var db = scope.ServiceProvider.GetRequiredService<BitweenDbContext>();
+
+        var doc = new Document(null, Unique("Agg config doc"), DocumentFormat.Json);
+        db.Set<Document>().Add(doc);
+        var partner = new Partner(Unique("Agg config partner"));
+        db.Set<Partner>().Add(partner);
+        await db.SaveChangesAsync();
+
+        var source = new Subscription(Unique("Agg config source"), doc.Id);
+        db.Set<Subscription>().Add(source);
+        await db.SaveChangesAsync();
+
+        return (source.Id, partner.Id);
+    }
+
+    private static ScheduleView[] Daily() =>
+        [new ScheduleView { Recurrence = Recurrence.Daily, Hours = 2 }];
+
+    private async Task<int> CreateAggregation(int sourceId, int partnerId, XchangeFileType? target)
+    {
+        await using var scope = _fixture.CreateScope();
+        scope.Superuser();
+        var handler = ActivatorUtilities.CreateInstance<Resources.Subscriptions.Create>(scope.ServiceProvider);
+
+        var model = new SubscriptionCreate
+        {
+            Name = Unique("Agg config"),
+            Type = SubscriptionType.Aggregation,
+            AggregationForId = sourceId,
+            PartnerId = partnerId,
+            Schedules = Daily(),
+        };
+        // Left unset on purpose by the caller testing the default, so that the test proves the
+        // default rather than the applier's willingness to copy whatever it was handed.
+        if (target.HasValue) model.AggregationTarget = target.Value;
+
+        return (int)await handler.Handle(model);
+    }
+
+    [Fact]
+    public async Task An_aggregation_can_be_created_collecting_the_mapped_file()
+    {
+        var (sourceId, partnerId) = await Groundwork();
+        var id = await CreateAggregation(sourceId, partnerId, XchangeFileType.Output);
+
+        await using var scope = _fixture.CreateScope();
+        var db = scope.ServiceProvider.GetRequiredService<BitweenDbContext>();
+        var entity = await db.Set<Subscription>().SingleAsync(s => s.Id == id);
+
+        Assert.Equal(XchangeFileType.Output, entity.AggregationTarget);
+    }
+
+    [Fact]
+    public async Task An_aggregation_collects_what_came_in_unless_told_otherwise()
+    {
+        var (sourceId, partnerId) = await Groundwork();
+        var id = await CreateAggregation(sourceId, partnerId, null);
+
+        await using var scope = _fixture.CreateScope();
+        var db = scope.ServiceProvider.GetRequiredService<BitweenDbContext>();
+        var entity = await db.Set<Subscription>().SingleAsync(s => s.Id == id);
+
+        Assert.Equal(XchangeFileType.Input, entity.AggregationTarget);
+    }
+
+    [Fact]
+    public async Task Changing_which_file_is_collected_is_kept()
+    {
+        var (sourceId, partnerId) = await Groundwork();
+        var id = await CreateAggregation(sourceId, partnerId, XchangeFileType.Input);
+
+        await using (var scope = _fixture.CreateScope())
+        {
+            scope.Superuser();
+            var update = ActivatorUtilities.CreateInstance<Resources.Subscriptions.Update>(scope.ServiceProvider);
+            await update.Handle(id, new SubscriptionUpdate
+            {
+                Name = Unique("Agg config renamed"),
+                AggregationForId = sourceId,
+                PartnerId = partnerId,
+                Schedules = Daily(),
+                AggregationTarget = XchangeFileType.Response,
+            });
+        }
+
+        await using var check = _fixture.CreateScope();
+        var db = check.ServiceProvider.GetRequiredService<BitweenDbContext>();
+        var entity = await db.Set<Subscription>().SingleAsync(s => s.Id == id);
+
+        Assert.Equal(XchangeFileType.Response, entity.AggregationTarget);
+    }
+
+    [Fact]
+    public async Task The_list_reports_when_an_aggregation_next_runs()
+    {
+        var (sourceId, partnerId) = await Groundwork();
+        var id = await CreateAggregation(sourceId, partnerId, XchangeFileType.Output);
+
+        await using var scope = _fixture.CreateScope();
+        scope.Superuser();
+        var search = ActivatorUtilities.CreateInstance<Resources.Subscriptions.Search>(scope.ServiceProvider);
+
+        var response = (SearchyResponse<SubscriptionSearch>)await search.Handle(
+            new SearchyRequest { PageSize = 500, PageIndex = 0 });
+        var row = response.Result.Single(r => r.Id == id);
+
+        // ReceiveOn is only ever set for Receiving, and the list page reads one next-run field
+        // for both scheduled types — so without AggregateOn in the projection every aggregation
+        // row showed no next run at all, for a job that plainly has one.
+        Assert.NotNull(row.AggregateOn);
+        Assert.Null(row.ReceiveOn);
+        Assert.Equal(XchangeFileType.Output, row.AggregationTarget);
+    }
 }

SW.Bitween.Sdk/Model/Subscription.cs

@@ -195,6 +195,17 @@ public abstract class SubscriptionConfiguration : SubscriptionCreateUpdateBase
 
         public int? RetryPolicyId { get; set; }
         public CustomRetryPolicy CustomRetryPolicy { get; set; }
+
+        /// <summary>
+        /// Aggregation only: which file of each collected exchange the roll-up links to.
+        /// <para>
+        /// Here rather than on <see cref="SubscriptionUpdate"/>, where it used to live, because
+        /// it is a choice a person makes and not runtime state. On update alone it reached the
+        /// entity through the generic property copy, so create could not set it and every
+        /// aggregation started on <see cref="XchangeFileType.Input"/> whatever was wanted.
+        /// </para>
+        /// </summary>
+        public XchangeFileType AggregationTarget { get; set; }
     }
 
     /// <summary>
@@ -244,7 +255,6 @@ public class SubscriptionUpdate : SubscriptionConfiguration
         public DateTime? AggregateOn { get; set; }
         public int ConsecutiveFailures { get; set; }
         public string LastException { get; set; }
-        public XchangeFileType AggregationTarget { get; set; }
         public DateTime? PausedOn { get; set; }
         public string CategoryCode { get; set; }
         public string CategoryDescription { get; set; }