Fix/traewelling full sweep

OV_DB.Tests/TraewellingResyncQueueTests.cs

@@ -0,0 +1,54 @@
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using OV_DB.Services;
+
+namespace OV_DB.Tests;
+
+/// <summary>
+/// A full-history walk takes minutes, so the button stays pressable long before the previous
+/// walk is done. Keeping one per user is what stops a second walk re-reading the same pages
+/// and racing the first on the inbox's unique index.
+/// </summary>
+public class TraewellingResyncQueueTests
+{
+    [Fact]
+    public void RefusesASecondResyncWhileOneIsPending()
+    {
+        var queue = new TraewellingResyncQueue();
+
+        Assert.True(queue.TryEnqueue(1));
+        Assert.False(queue.TryEnqueue(1));
+        Assert.True(queue.IsRunning(1));
+    }
+
+    [Fact]
+    public void LetsAnotherUserThrough()
+    {
+        var queue = new TraewellingResyncQueue();
+        queue.TryEnqueue(1);
+
+        Assert.True(queue.TryEnqueue(2));
+        Assert.False(queue.IsRunning(3));
+    }
+
+    [Fact]
+    public async Task AcceptsTheUserAgainOnceTheirWalkFinished()
+    {
+        var queue = new TraewellingResyncQueue();
+        queue.TryEnqueue(1);
+
+        var dequeued = await FirstAsync(queue);
+        queue.MarkFinished(dequeued);
+
+        Assert.False(queue.IsRunning(1));
+        Assert.True(queue.TryEnqueue(1));
+    }
+
+    private static async Task<int> FirstAsync(TraewellingResyncQueue queue)
+    {
+        await foreach (var userId in queue.DequeueAllAsync(CancellationToken.None))
+            return userId;
+        return 0;
+    }
+}

OV_DB.Tests/TrawellingInboxSweepTests.cs

@@ -0,0 +1,187 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Net;
+using System.Net.Http;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.AspNetCore.SignalR;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.Caching.Memory;
+using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.Logging.Abstractions;
+using Moq;
+using OV_DB.Hubs;
+using OV_DB.Services;
+using OVDB_database.Database;
+using OVDB_database.Models;
+
+namespace OV_DB.Tests;
+
+/// <summary>
+/// The statuses listing is ordered newest-first, so a check-in that was missed while newer ones
+/// were already imported sits behind a page holding nothing new. Which sweep mode walks past
+/// such a page decides whether it can ever be found again.
+/// </summary>
+public class TrawellingInboxSweepTests
+{
+    // Only the fields the sweep reads: the id it dedupes on and the departure it sorts and
+    // bounds the delete-healing by.
+    private static string Page(string next, params (int Id, string Departure)[] statuses)
+    {
+        var data = string.Join(",", statuses.Select(s =>
+            $$$"""{"id":{{{s.Id}}},"checkin":{"manualDeparture":"{{{s.Departure}}}"}}"""));
+        var nextLink = next == null ? "null" : $"\"{next}\"";
+        return $$$"""{"data":[{{{data}}}],"links":{"next":{{{nextLink}}}}}""";
+    }
+
+    // Progress<T> hands its reports to the thread pool, which would make the order they are
+    // asserted in a race.
+    private sealed class InlineProgress<T>(Action<T> report) : IProgress<T>
+    {
+        public void Report(T value) => report(value);
+    }
+
+    private sealed class PagedHandler(Dictionary<int, string> pages) : HttpMessageHandler
+    {
+        public List<int> RequestedPages { get; } = [];
+
+        protected override Task<HttpResponseMessage> SendAsync(HttpRequestMessage request, CancellationToken cancellationToken)
+        {
+            var page = int.Parse(request.RequestUri!.Query.Split("page=")[1]);
+            RequestedPages.Add(page);
+            return Task.FromResult(new HttpResponseMessage(HttpStatusCode.OK)
+            {
+                Content = new StringContent(pages[page])
+            });
+        }
+    }
+
+    private static (TrawellingService Service, User User, OVDBDatabaseContext Context, PagedHandler Handler) NewService(
+        Dictionary<int, string> pages)
+    {
+        var handler = new PagedHandler(pages);
+        var factory = new Mock<IHttpClientFactory>();
+        factory.Setup(f => f.CreateClient(It.IsAny<string>()))
+            .Returns(() => new HttpClient(handler) { BaseAddress = new Uri("https://traewelling.test/") });
+
+        var configuration = new ConfigurationBuilder()
+            .AddInMemoryCollection(new Dictionary<string, string>
+            {
+                ["Traewelling:BaseUrl"] = "https://traewelling.test/api/v1"
+            })
+            .Build();
+
+        var options = new DbContextOptionsBuilder<OVDBDatabaseContext>()
+            .UseInMemoryDatabase(Guid.NewGuid().ToString())
+            .Options;
+        var context = new OVDBDatabaseContext(options);
+
+        var service = new TrawellingService(
+            factory.Object,
+            configuration,
+            new Mock<ITimezoneService>().Object,
+            context,
+            NullLogger<TrawellingService>.Instance,
+            new MemoryCache(new MemoryCacheOptions()),
+            Mock.Of<ITraewellingRateLimiter>(),
+            new Mock<IHubContext<TraewellingHub>>().Object);
+
+        var user = new User
+        {
+            Id = 1,
+            // Set, so the sweep doesn't first call /auth/user to look it up.
+            TrawellingUsername = "gertrud",
+            TrawellingAccessToken = "token",
+            TrawellingRefreshToken = "refresh",
+            TrawellingTokenExpiresAt = DateTime.UtcNow.AddHours(1)
+        };
+
+        return (service, user, context, handler);
+    }
+
+    // Page 1 holds two check-ins already in the inbox, page 2 an older one that never arrived.
+    private static Dictionary<int, string> NewestKnownOldestMissing() => new()
+    {
+        [1] = Page("https://traewelling.test/api/v1/user/gertrud/statuses?page=2",
+            (101, "2026-08-20T10:00:00+02:00"),
+            (102, "2026-08-19T10:00:00+02:00")),
+        [2] = Page(null, (50, "2026-06-01T10:00:00+02:00")),
+    };
+
+    private static void SeedInbox(OVDBDatabaseContext context, params (int StatusId, string Departure)[] statuses)
+    {
+        foreach (var (statusId, departure) in statuses)
+        {
+            context.TrawellingInboxStatuses.Add(new TrawellingInboxStatus
+            {
+                UserId = 1,
+                TrawellingStatusId = statusId,
+                PayloadJson = "{}",
+                State = TrawellingInboxState.Pending,
+                Source = TrawellingInboxSource.Sweep,
+                DepartureAt = DateTimeOffset.Parse(departure).UtcDateTime,
+                ReceivedAt = DateTime.UtcNow,
+                LastEventAt = DateTime.UtcNow,
+            });
+        }
+        context.SaveChanges();
+    }
+
+    [Fact]
+    public async Task ForcedSweepStopsAtTheFirstPageWithoutAnythingNew()
+    {
+        var (service, user, context, handler) = NewService(NewestKnownOldestMissing());
+        SeedInbox(context, (101, "2026-08-20T10:00:00+02:00"), (102, "2026-08-19T10:00:00+02:00"));
+
+        var result = await service.SweepInboxAsync(user, TrawellingSweepMode.Force);
+
+        Assert.True(result.Success);
+        Assert.Equal(0, result.Added);
+        Assert.Equal([1], handler.RequestedPages);
+        Assert.DoesNotContain(50, context.TrawellingInboxStatuses.Select(s => s.TrawellingStatusId));
+    }
+
+    [Fact]
+    public async Task FullSweepWalksPastKnownPagesAndPicksUpTheMissedCheckIn()
+    {
+        var (service, user, context, handler) = NewService(NewestKnownOldestMissing());
+        SeedInbox(context, (101, "2026-08-20T10:00:00+02:00"), (102, "2026-08-19T10:00:00+02:00"));
+        var reported = new List<TrawellingSweepProgress>();
+
+        var result = await service.SweepInboxAsync(user, TrawellingSweepMode.Full,
+            new InlineProgress<TrawellingSweepProgress>(reported.Add));
+
+        Assert.True(result.Success);
+        Assert.True(result.ReachedEnd);
+        Assert.Equal(1, result.Added);
+        Assert.Equal([1, 2], handler.RequestedPages);
+        Assert.Contains(50, context.TrawellingInboxStatuses.Select(s => s.TrawellingStatusId));
+        // One report per page, so the page it is on is visible while the walk runs.
+        Assert.Equal([(1, 0), (2, 1)], reported.Select(p => (p.PagesRead, p.Added)));
+    }
+
+    [Fact]
+    public async Task FullSweepKeepsStatusesAlreadyKnownFromTheSweptRange()
+    {
+        var (service, user, context, _) = NewService(NewestKnownOldestMissing());
+        SeedInbox(context, (101, "2026-08-20T10:00:00+02:00"), (102, "2026-08-19T10:00:00+02:00"));
+
+        await service.SweepInboxAsync(user, TrawellingSweepMode.Full);
+
+        Assert.Equal([50, 101, 102], context.TrawellingInboxStatuses
+            .Select(s => s.TrawellingStatusId).OrderBy(id => id));
+    }
+
+    [Fact]
+    public async Task StaleCheckSkipsTheApiEntirelyWhenTheLastSweepIsRecent()
+    {
+        var (service, user, _, handler) = NewService(NewestKnownOldestMissing());
+        user.TrawellingLastSweepAt = DateTime.UtcNow.AddMinutes(-5);
+
+        var result = await service.SweepInboxAsync(user);
+
+        Assert.True(result.Success);
+        Assert.Empty(handler.RequestedPages);
+    }
+}

OV_DB/Controllers/TraewellingController.cs

@@ -28,14 +28,17 @@ public class TraewellingController : ControllerBase
         private readonly OVDBDatabaseContext _dbContext;
         private readonly ILogger<TraewellingController> _logger;
         private readonly ITraewellingRateLimiter _rateLimiter;
+        private readonly ITraewellingResyncQueue _resyncQueue;
 
         public TraewellingController(ITrawellingService trawellingService,
-            OVDBDatabaseContext dbContext, ILogger<TraewellingController> logger, ITraewellingRateLimiter rateLimiter)
+            OVDBDatabaseContext dbContext, ILogger<TraewellingController> logger, ITraewellingRateLimiter rateLimiter,
+            ITraewellingResyncQueue resyncQueue)
         {
             _trawellingService = trawellingService;
             _dbContext = dbContext;
             _logger = logger;
             _rateLimiter = rateLimiter;
+            _resyncQueue = resyncQueue;
         }
 
         /// <summary>
@@ -268,7 +271,8 @@ public async Task<IActionResult> GetConnectionStatus()
                     user = userInfo,
                     liveSyncAvailable = _trawellingService.IsLiveSyncAvailable,
                     liveSyncEnabled = user.TraewellingWebhookId.HasValue,
-                    liveSyncHealth = webhookHealth.ToString()
+                    liveSyncHealth = webhookHealth.ToString(),
+                    resyncRunning = _resyncQueue.IsRunning(user.Id)
                 });
             }
             catch (Exception ex)
@@ -314,6 +318,37 @@ public async Task<IActionResult> GetUnimportedTrips([FromQuery] int page = 1, [F
             }
         }
 
+        /// <summary>
+        /// Queue a walk of the complete Träwelling history to pick up check-ins the regular
+        /// sweep stops short of: ones older than check-ins that were already imported.
+        /// Walking a whole history takes minutes, so the walk itself runs on
+        /// <see cref="TraewellingResyncService"/> and reports over the hub.
+        /// </summary>
+        [HttpPost("resync")]
+        public async Task<IActionResult> ResyncFullHistory()
+        {
+            try
+            {
+                var userId = GetCurrentUserId();
+                if (!userId.HasValue)
+                    return Unauthorized();
+
+                var user = await _dbContext.Users.FindAsync(userId.Value);
+                if (user == null)
+                    return NotFound("User not found");
+
+                if (!_trawellingService.IsConnected(user))
+                    return BadRequest("Träwelling account not connected or tokens expired");
+
+                return Accepted(new { queued = _resyncQueue.TryEnqueue(userId.Value) });
+            }
+            catch (Exception ex)
+            {
+                _logger.LogError(ex, "Error queueing Träwelling resync");
+                return StatusCode(500, "Error queueing Träwelling resync");
+            }
+        }
+
         /// <summary>
         /// Ignore a Träwelling status so it doesn't show up in unimported list
         /// </summary>

OV_DB/Hubs/TraewellingHub.cs

@@ -4,9 +4,10 @@
 namespace OV_DB.Hubs;
 
 /// <summary>
-/// Per-user live updates for the Träwelling unimported-trips list, fed by webhook events.
-/// Authenticated with the regular JWT (access_token query parameter on the handshake);
-/// events go through Clients.User so they only reach the owning user's connections.
+/// Per-user live updates for the Träwelling unimported-trips list, fed by webhook events
+/// and by full-history resyncs. Authenticated with the regular JWT (access_token query
+/// parameter on the handshake); events go through Clients.User so they only reach the
+/// owning user's connections.
 /// </summary>
 [Authorize]
 public class TraewellingHub : Hub
@@ -18,4 +19,9 @@ public class TraewellingHub : Hub
     public const string PendingTripRemovedMethod = "PendingTripRemoved";
     public const string ConflictUpsertedMethod = "ConflictUpserted";
     public const string ConflictRemovedMethod = "ConflictRemoved";
+
+    // A resync outlives the request that asked for it, so its progress and its outcome
+    // reach the page only here.
+    public const string ResyncProgressMethod = "ResyncProgress";
+    public const string ResyncFinishedMethod = "ResyncFinished";
 }

OV_DB/OVDBFrontend/src/app/models/traewelling.model.ts

@@ -8,6 +8,8 @@ export interface TrawellingConnectionStatus {
   liveSyncAvailable?: boolean;
   liveSyncEnabled?: boolean;
   liveSyncHealth?: 'NotEnabled' | 'Active' | 'DisabledUpstream' | 'Missing' | 'Unknown';
+  /** A full-history resync started earlier is still walking; its finish still arrives over the hub. */
+  resyncRunning?: boolean;
 }
 
 export interface TrawellingUser {
@@ -158,6 +160,24 @@ export interface TrawellingIgnoreResponse {
   message?: string;
 }
 
+/** 202 response of the resync endpoint: false when one was already running for this user. */
+export interface TrawellingResyncQueued {
+  queued: boolean;
+}
+
+/** Pushed over the hub every few pages while the walk runs. */
+export interface TrawellingResyncProgress {
+  pagesRead: number;
+  added: number;
+}
+
+export interface TrawellingResyncResult {
+  added: number;
+  pagesRead: number;
+  /** False when the walk stopped on an error or the page limit, so older check-ins may remain unseen. */
+  complete: boolean;
+}
+
 export interface RoutesListResponse {
   count: number;
   routes: RouteSearchResult[];