From 334e0520579215261268a9c9ae0249c057cf5fb3 Mon Sep 17 00:00:00 2001 From: Jaap Slootbeek Date: Thu, 17 Sep 2026 15:58:08 +0200 Subject: [PATCH 1/2] fix(traewelling): reach check-ins the sweep stops short of A check-in that never reached the inbox stayed unreachable once newer ones had been imported. The sweep walks /user/{username}/statuses newest-first and breaks on the first page where nothing is new, and "new" is measured against everything known: imported, ignored, and already pending. Once the newest page is entirely known - the steady state with live sync on - the walk ends at page 1, and anything missed that has scrolled past it is never asked for again. The refresh button did not help: it only bypassed the one-hour staleness check, not the break. Nor did paging the list, which pages the local inbox rather than upstream. So the sweep gains a mode. WhenStale and Force keep the break and now read up to ten pages instead of five; Full ignores it and follows links.next to the end, behind a new history button next to refresh and POST /api/traewelling/resync. Its page limit is a safety valve at 1000, not a depth to tune - a rerun starts at page 1 again, so there is nothing to page past - and the result says whether the walk ran out of pages or stopped short, which the frontend passes on. Two things make a long walk survivable. Each page saves before the next is asked for, so a walk cut off midway still leaves the inbox fuller than it was, and the endpoint deliberately does not pass the request's cancellation token: a proxy timeout or an impatient browser should not throw away half an hour of paging. The delete-healing needed no change. It is bounded by the oldest departure the sweep actually saw, and Traewelling orders this listing by departure descending (statusesForUser: orderByDesc on train_checkins.departure, 15 per page), so the bound covers exactly the pages that were read whether the walk was cut short or not. The tests pin the behaviour that was wrong: against a fixture whose first page is entirely known and whose second holds an older missed check-in, Force stops after one request and finds nothing, Full reads both and picks it up. Co-Authored-By: Claude Opus 5 (1M context) --- OV_DB.Tests/TrawellingInboxSweepTests.cs | 176 ++++++++++++++++++ OV_DB/Controllers/TraewellingController.cs | 41 ++++ .../src/app/models/traewelling.model.ts | 7 + .../services/traewelling.service.ts | 9 +- .../traewelling/traewelling.component.html | 4 + .../traewelling/traewelling.component.scss | 4 + .../app/traewelling/traewelling.component.ts | 23 +++ OV_DB/OVDBFrontend/src/assets/i18n/en.json | 4 + OV_DB/OVDBFrontend/src/assets/i18n/nl.json | 4 + OV_DB/Services/ITrawellingService.cs | 32 +++- .../Services/TraewellingInboxSweepService.cs | 2 +- OV_DB/Services/TrawellingService.cs | 56 ++++-- docs/traewelling-integration.md | 25 ++- 13 files changed, 362 insertions(+), 25 deletions(-) create mode 100644 OV_DB.Tests/TrawellingInboxSweepTests.cs diff --git a/OV_DB.Tests/TrawellingInboxSweepTests.cs b/OV_DB.Tests/TrawellingInboxSweepTests.cs new file mode 100644 index 00000000..82cc3876 --- /dev/null +++ b/OV_DB.Tests/TrawellingInboxSweepTests.cs @@ -0,0 +1,176 @@ +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; + +/// +/// 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. +/// +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}}}}}"""; + } + + private sealed class PagedHandler(Dictionary pages) : HttpMessageHandler + { + public List RequestedPages { get; } = []; + + protected override Task 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 pages) + { + var handler = new PagedHandler(pages); + var factory = new Mock(); + factory.Setup(f => f.CreateClient(It.IsAny())) + .Returns(() => new HttpClient(handler) { BaseAddress = new Uri("https://traewelling.test/") }); + + var configuration = new ConfigurationBuilder() + .AddInMemoryCollection(new Dictionary + { + ["Traewelling:BaseUrl"] = "https://traewelling.test/api/v1" + }) + .Build(); + + var options = new DbContextOptionsBuilder() + .UseInMemoryDatabase(Guid.NewGuid().ToString()) + .Options; + var context = new OVDBDatabaseContext(options); + + var service = new TrawellingService( + factory.Object, + configuration, + new Mock().Object, + context, + NullLogger.Instance, + new MemoryCache(new MemoryCacheOptions()), + Mock.Of(), + new Mock>().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 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 result = await service.SweepInboxAsync(user, TrawellingSweepMode.Full); + + 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)); + } + + [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); + } +} diff --git a/OV_DB/Controllers/TraewellingController.cs b/OV_DB/Controllers/TraewellingController.cs index f7627a55..62e553ea 100644 --- a/OV_DB/Controllers/TraewellingController.cs +++ b/OV_DB/Controllers/TraewellingController.cs @@ -314,6 +314,47 @@ public async Task GetUnimportedTrips([FromQuery] int page = 1, [F } } + /// + /// Walk 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. + /// + [HttpPost("resync")] + public async Task 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"); + + // Deliberately not the request's token: a long walk that outlives an impatient + // browser or a proxy timeout should still finish and keep what it found. + var result = await _trawellingService.SweepInboxAsync(user, TrawellingSweepMode.Full); + + if (!result.Success && _rateLimiter.IsLimited) + return StatusCode(429, "Träwelling is rate limiting us, please try again in a minute"); + + return Ok(new + { + added = result.Added, + pagesRead = result.PagesRead, + complete = result.Success && result.ReachedEnd + }); + } + catch (Exception ex) + { + _logger.LogError(ex, "Error resyncing Träwelling history"); + return StatusCode(500, "Error resyncing Träwelling history"); + } + } + /// /// Ignore a Träwelling status so it doesn't show up in unimported list /// diff --git a/OV_DB/OVDBFrontend/src/app/models/traewelling.model.ts b/OV_DB/OVDBFrontend/src/app/models/traewelling.model.ts index 8a3e472d..a2f1c246 100644 --- a/OV_DB/OVDBFrontend/src/app/models/traewelling.model.ts +++ b/OV_DB/OVDBFrontend/src/app/models/traewelling.model.ts @@ -158,6 +158,13 @@ export interface TrawellingIgnoreResponse { message?: string; } +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[]; diff --git a/OV_DB/OVDBFrontend/src/app/traewelling/services/traewelling.service.ts b/OV_DB/OVDBFrontend/src/app/traewelling/services/traewelling.service.ts index fbb19a3c..d34796ba 100644 --- a/OV_DB/OVDBFrontend/src/app/traewelling/services/traewelling.service.ts +++ b/OV_DB/OVDBFrontend/src/app/traewelling/services/traewelling.service.ts @@ -18,7 +18,8 @@ import { RoutesListResponse, TraewellingAlert, TrawellingConflict, - TrawellingConflictAction + TrawellingConflictAction, + TrawellingResyncResult } from '../../models/traewelling.model'; import { TraewellingTagMapping } from '../../models/user-profile.model'; import { ApiService } from '../../services/api.service'; @@ -58,6 +59,12 @@ export class TrawellingService { ); } + async resyncFullHistory(): Promise { + return firstValueFrom(this.http.post(`${this.baseUrl}/resync`, {}) + .pipe(first()) + ); + } + async getConflicts(): Promise { return firstValueFrom(this.http.get(`${this.baseUrl}/conflicts`) .pipe(first()) diff --git a/OV_DB/OVDBFrontend/src/app/traewelling/traewelling.component.html b/OV_DB/OVDBFrontend/src/app/traewelling/traewelling.component.html index 47d0c495..23222360 100644 --- a/OV_DB/OVDBFrontend/src/app/traewelling/traewelling.component.html +++ b/OV_DB/OVDBFrontend/src/app/traewelling/traewelling.component.html @@ -18,6 +18,10 @@ [attr.aria-label]="'TRAEWELLING.REFRESH' | translate"> refresh + } diff --git a/OV_DB/OVDBFrontend/src/app/traewelling/traewelling.component.scss b/OV_DB/OVDBFrontend/src/app/traewelling/traewelling.component.scss index e4b4adbc..447fbdcb 100644 --- a/OV_DB/OVDBFrontend/src/app/traewelling/traewelling.component.scss +++ b/OV_DB/OVDBFrontend/src/app/traewelling/traewelling.component.scss @@ -30,6 +30,10 @@ align-self: center; } + .resync-button { + align-self: center; + } + .conflicts-container { margin-bottom: 16px; diff --git a/OV_DB/OVDBFrontend/src/app/traewelling/traewelling.component.ts b/OV_DB/OVDBFrontend/src/app/traewelling/traewelling.component.ts index 71acb8be..118db627 100644 --- a/OV_DB/OVDBFrontend/src/app/traewelling/traewelling.component.ts +++ b/OV_DB/OVDBFrontend/src/app/traewelling/traewelling.component.ts @@ -54,6 +54,7 @@ export class TrawellingComponent implements OnInit, OnDestroy { conflicts: TrawellingConflict[] = []; alerts: TraewellingAlert[] = []; isLoading = true; + isResyncing = false; isLoadingMore = false; hasMorePages = false; currentPage = 1; @@ -202,6 +203,28 @@ export class TrawellingComponent implements OnInit, OnDestroy { } } + async resyncFullHistory() { + if (this.isResyncing) return; + this.isResyncing = true; + try { + const result = await this.trawellingService.resyncFullHistory(); + const messageKey = !result.complete + ? 'TRAEWELLING.RESYNC_INCOMPLETE' + : result.added > 0 ? 'TRAEWELLING.RESYNC_FOUND' : 'TRAEWELLING.RESYNC_NOTHING_FOUND'; + this.snackBar.open( + this.translateService.instant(messageKey, { count: result.added }), + this.translateService.instant('CLOSE'), + { duration: 5000 }); + if (result.added > 0) { + await this.loadTrips(); + } + } catch { + // The global error interceptor already showed a toast + } finally { + this.isResyncing = false; + } + } + private async loadTrips(refresh = false) { try { const response = await this.trawellingService.getUnimportedTrips(1, refresh); diff --git a/OV_DB/OVDBFrontend/src/assets/i18n/en.json b/OV_DB/OVDBFrontend/src/assets/i18n/en.json index 374d13c4..c2c0b087 100644 --- a/OV_DB/OVDBFrontend/src/assets/i18n/en.json +++ b/OV_DB/OVDBFrontend/src/assets/i18n/en.json @@ -728,6 +728,10 @@ "CATEGORY_TAXI": "Taxi", "CATEGORY_UNKNOWN": "Unknown", "REFRESH": "Check Träwelling for new trips", + "RESYNC": "Search your whole Träwelling history for trips that were skipped", + "RESYNC_FOUND": "Skipped trips found: {{count}}", + "RESYNC_NOTHING_FOUND": "No skipped trips found, your whole Träwelling history is here", + "RESYNC_INCOMPLETE": "Could not read your whole history, try again. Trips found so far: {{count}}", "IGNORE_TRIP": "Ignore trip", "IGNORE_CONFIRM": "ignore this trip (it will not come back in this list)", "IGNORING": "Ignoring trip...", diff --git a/OV_DB/OVDBFrontend/src/assets/i18n/nl.json b/OV_DB/OVDBFrontend/src/assets/i18n/nl.json index 73341214..9cbdb777 100644 --- a/OV_DB/OVDBFrontend/src/assets/i18n/nl.json +++ b/OV_DB/OVDBFrontend/src/assets/i18n/nl.json @@ -726,6 +726,10 @@ "CATEGORY_TAXI": "Taxi", "CATEGORY_UNKNOWN": "Onbekend", "REFRESH": "Träwelling controleren op nieuwe reizen", + "RESYNC": "Je hele Träwelling-geschiedenis doorzoeken op overgeslagen reizen", + "RESYNC_FOUND": "Overgeslagen reizen gevonden: {{count}}", + "RESYNC_NOTHING_FOUND": "Geen overgeslagen reizen gevonden, je hele Träwelling-geschiedenis staat er al in", + "RESYNC_INCOMPLETE": "Kon je hele geschiedenis niet lezen, probeer het opnieuw. Tot nu toe gevonden: {{count}}", "IGNORE_TRIP": "Reis negeren", "IGNORE_CONFIRM": "deze reis wilt negeren (hij komt niet terug in deze lijst)", "IGNORING": "Reis negeren...", diff --git a/OV_DB/Services/ITrawellingService.cs b/OV_DB/Services/ITrawellingService.cs index 084937a4..786d6db7 100644 --- a/OV_DB/Services/ITrawellingService.cs +++ b/OV_DB/Services/ITrawellingService.cs @@ -7,6 +7,30 @@ namespace OV_DB.Services { + /// + /// How far back a sweep reads the Träwelling statuses listing. + /// + public enum TrawellingSweepMode + { + /// Skip entirely while the last sweep is still fresh. + WhenStale, + + /// Sweep now, stopping at the first page that holds nothing new. + Force, + + /// + /// Walk the whole listing, past pages that hold nothing new. The only way to reach a + /// check-in that was missed while newer ones were already imported. + /// + Full, + } + + /// False when the listing could not be read. + /// Statuses newly added to the inbox. + /// Pages of the statuses listing that were fetched. + /// True when the walk ran out of pages rather than stopping at a cap. + public record TrawellingSweepResult(bool Success, int Added, int PagesRead, bool ReachedEnd); + public interface ITrawellingService { /// @@ -75,14 +99,12 @@ public interface ITrawellingService /// /// Reconcile the local inbox with the Träwelling statuses API: adds unknown statuses - /// as pending, removes pending rows that were deleted upstream. Skipped when the last - /// sweep is fresh, unless forced. + /// as pending, removes pending rows that were deleted upstream. /// /// User to sweep for - /// Sweep even when the last sweep is recent + /// How far back to read the listing /// Cancellation token - /// True when the sweep completed (or was fresh enough to skip) - Task SweepInboxAsync(User user, bool force = false, System.Threading.CancellationToken cancellationToken = default); + Task SweepInboxAsync(User user, TrawellingSweepMode mode = TrawellingSweepMode.WhenStale, System.Threading.CancellationToken cancellationToken = default); /// /// Remove a status from the inbox after it has been imported or ignored diff --git a/OV_DB/Services/TraewellingInboxSweepService.cs b/OV_DB/Services/TraewellingInboxSweepService.cs index 133ce0bc..9aa2cbbc 100644 --- a/OV_DB/Services/TraewellingInboxSweepService.cs +++ b/OV_DB/Services/TraewellingInboxSweepService.cs @@ -55,7 +55,7 @@ private async Task SweepAllUsersAsync(CancellationToken cancellationToken) foreach (var user in connectedUsers) { cancellationToken.ThrowIfCancellationRequested(); - if (await trawellingService.SweepInboxAsync(user, force: true, cancellationToken)) + if ((await trawellingService.SweepInboxAsync(user, TrawellingSweepMode.Force, cancellationToken)).Success) swept++; // Daily job, so latency is free — don't burst through the shared rate limit await Task.Delay(TimeSpan.FromSeconds(5), cancellationToken); diff --git a/OV_DB/Services/TrawellingService.cs b/OV_DB/Services/TrawellingService.cs index cabcbc89..78db5c6b 100644 --- a/OV_DB/Services/TrawellingService.cs +++ b/OV_DB/Services/TrawellingService.cs @@ -321,21 +321,28 @@ public async Task GetUserInfoAsync(User user) } } - private const int SweepMaxPages = 5; + private const int SweepMaxPages = 10; + + // Safety valve for the full walk rather than a depth to tune: a rerun starts at page 1 + // again, so hitting this is not something the user can page past. + private const int FullSweepMaxPages = 1000; + private static readonly TimeSpan SweepStaleness = TimeSpan.FromHours(1); - public async Task SweepInboxAsync(User user, bool force = false, CancellationToken cancellationToken = default) + public async Task SweepInboxAsync(User user, TrawellingSweepMode mode = TrawellingSweepMode.WhenStale, CancellationToken cancellationToken = default) { + var added = 0; + var pagesRead = 0; try { - if (!force && user.TrawellingLastSweepAt.HasValue && + if (mode == TrawellingSweepMode.WhenStale && user.TrawellingLastSweepAt.HasValue && DateTime.UtcNow - user.TrawellingLastSweepAt.Value < SweepStaleness) { - return true; + return new TrawellingSweepResult(true, 0, 0, false); } if (!await EnsureValidTokenAsync(user)) - return false; + return new TrawellingSweepResult(false, 0, 0, false); // Ensure we have the username for this user if (string.IsNullOrEmpty(user.TrawellingUsername)) @@ -344,7 +351,7 @@ public async Task SweepInboxAsync(User user, bool force = false, Cancellat if (userInfo == null) { _logger.LogError("Could not fetch Träwelling username for user {UserId}", user.Id); - return false; + return new TrawellingSweepResult(false, 0, 0, false); } user.TrawellingUsername = userInfo.Username; await _dbContext.SaveChangesAsync(cancellationToken); @@ -370,9 +377,10 @@ public async Task SweepInboxAsync(User user, bool force = false, Cancellat var seenIds = new HashSet(); DateTime? oldestSweptDeparture = null; - var added = 0; + var maxPages = mode == TrawellingSweepMode.Full ? FullSweepMaxPages : SweepMaxPages; + var reachedEnd = false; - for (var page = 1; page <= SweepMaxPages; page++) + for (var page = 1; page <= maxPages; page++) { var response = await SendAsync(() => CreateApiRequest(HttpMethod.Get, $"{_baseUrl}/user/{user.TrawellingUsername}/statuses?page={page}", user), @@ -381,13 +389,17 @@ public async Task SweepInboxAsync(User user, bool force = false, Cancellat if (!response.IsSuccessStatusCode) { _logger.LogError("Inbox sweep failed for user {UserId}. Status: {StatusCode}", user.Id, response.StatusCode); - return false; + return new TrawellingSweepResult(false, added, pagesRead, false); } + pagesRead++; var responseContent = await response.Content.ReadAsStringAsync(cancellationToken); var statusesResponse = JsonConvert.DeserializeObject(responseContent); if (statusesResponse?.Data == null || statusesResponse.Data.Count == 0) + { + reachedEnd = true; break; + } var newOnPage = 0; foreach (var status in statusesResponse.Data) @@ -416,12 +428,26 @@ public async Task SweepInboxAsync(User user, bool force = false, Cancellat added++; } - // Statuses are ordered newest-first: a full page without anything new means - // everything older is already known too. - if (newOnPage == 0 || string.IsNullOrEmpty(statusesResponse.Links?.Next)) + // Keep what this page found before asking for the next one, so a walk that is + // cut short still leaves the inbox fuller than it was. + if (newOnPage > 0) + await _dbContext.SaveChangesAsync(cancellationToken); + + if (string.IsNullOrEmpty(statusesResponse.Links?.Next)) + { + reachedEnd = true; + break; + } + + // Statuses are ordered newest-first, so a page without anything new only + // means everything older is known too while nothing was ever missed. + if (newOnPage == 0 && mode != TrawellingSweepMode.Full) break; } + if (mode == TrawellingSweepMode.Full && !reachedEnd) + _logger.LogWarning("Full inbox sweep for user {UserId} stopped at the {MaxPages}-page limit", user.Id, FullSweepMaxPages); + // Heal upstream deletes: a pending row inside the swept departure range that no // longer appears in the listing was deleted on Träwelling. Nothing was curated // yet, so it can simply be dropped. @@ -445,7 +471,7 @@ public async Task SweepInboxAsync(User user, bool force = false, Cancellat if (added > 0) _logger.LogInformation("Inbox sweep added {Added} statuses for user {UserId}", added, user.Id); - return true; + return new TrawellingSweepResult(true, added, pagesRead, reachedEnd); } catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { @@ -454,7 +480,7 @@ public async Task SweepInboxAsync(User user, bool force = false, Cancellat catch (Exception ex) { _logger.LogError(ex, "Error sweeping Träwelling inbox for user {UserId}", user.Id); - return false; + return new TrawellingSweepResult(false, added, pagesRead, false); } } @@ -473,7 +499,7 @@ public async Task GetOptimizedTripsAsync(User user, int // The inbox is the source of the list; the statuses API is only touched by the // sweep (skipped when fresh, forced by the frontend's refresh action). A failed // sweep still returns the current — possibly stale — inbox contents. - await SweepInboxAsync(user, force: refresh, cancellationToken); + await SweepInboxAsync(user, refresh ? TrawellingSweepMode.Force : TrawellingSweepMode.WhenStale, cancellationToken); const int pageSize = 15; var query = _dbContext.TrawellingInboxStatuses diff --git a/docs/traewelling-integration.md b/docs/traewelling-integration.md index 00127318..a2f86037 100644 --- a/docs/traewelling-integration.md +++ b/docs/traewelling-integration.md @@ -85,7 +85,13 @@ Returns connection status and Träwelling user info if connected. ``` ### GET /api/traewelling/unimported?page=1 -Returns paginated list of Träwelling trips not yet imported to OVDB. +Returns paginated list of Träwelling trips not yet imported to OVDB. Reads the local inbox; +`refresh=true` sweeps the statuses API first instead of waiting for the hourly staleness check. + +The regular sweep reads at most 10 pages and stops at the first page holding nothing new, +which is correct only while check-ins arrive in order. One that was missed — a webhook that +never landed, a check-in made before the account was connected — sits behind pages of +already-imported ones and is out of its reach. ### POST /api/traewelling/import Imports a specific Träwelling trip as an OVDB RouteInstance. @@ -99,8 +105,21 @@ Imports a specific Träwelling trip as an OVDB RouteInstance. } ``` -### POST /api/traewelling/process-backlog?maxPages=5 -Processes multiple pages of Träwelling history to import trips in bulk and enhance existing RouteInstances with timing data. +### POST /api/traewelling/resync +Walks the whole statuses listing, past pages holding nothing new, and adds everything still +unknown to the inbox. The history button next to refresh on the Träwelling page. + +**Response:** +```json +{ + "added": 3, + "pagesRead": 27, + "complete": true +} +``` + +`complete: false` means the walk stopped on an error or the 1000-page safety limit, so older +check-ins may still be missing; a rerun starts at page 1 again. ### GET /api/traewelling/stats Returns statistics about the user's Träwelling integration. From 25ee3e6d76b0b79b0c63608f798654e7c25d59d3 Mon Sep 17 00:00:00 2001 From: Jaap Slootbeek Date: Thu, 17 Sep 2026 16:16:19 +0200 Subject: [PATCH 2/2] feat(traewelling): run the full-history resync detached from the request Walking a whole history takes minutes, well past any reverse proxy's read timeout. Awaiting it in the request meant the browser got a 504 and a generic error toast for a sweep that was in fact still running and still saving pages - and the natural response to that toast is to press the button again. So the endpoint only queues now and answers 202. TraewellingResyncService does the walking off a channel, one walk per user at a time, and the page hears about it over the hub it already had open for webhook events: ResyncProgress every five pages, ResyncFinished with the totals. The page connects to the hub whenever the account is connected rather than only with live sync on, since without it there would be nothing to hear the resync on. GET /status gained resyncRunning, so a page opened or reloaded midway shows the walk rather than an idle button, and its finish still arrives over the hub. Progress comes out of the sweep as IProgress, reported after each page. Neither end uses Progress: it posts reports to the thread pool, which reorders them and races the counter the service throttles on. Co-Authored-By: Claude Opus 5 (1M context) --- OV_DB.Tests/TraewellingResyncQueueTests.cs | 54 +++++++++ OV_DB.Tests/TrawellingInboxSweepTests.cs | 13 +- OV_DB/Controllers/TraewellingController.cs | 32 ++--- OV_DB/Hubs/TraewellingHub.cs | 12 +- .../src/app/models/traewelling.model.ts | 13 ++ .../services/traewelling-live.service.ts | 28 ++++- .../services/traewelling.service.ts | 7 +- .../traewelling/traewelling.component.html | 12 +- .../traewelling/traewelling.component.scss | 6 + .../app/traewelling/traewelling.component.ts | 46 +++++--- OV_DB/OVDBFrontend/src/assets/i18n/en.json | 3 + OV_DB/OVDBFrontend/src/assets/i18n/nl.json | 3 + OV_DB/Services/ITrawellingService.cs | 6 +- .../Services/TraewellingInboxSweepService.cs | 2 +- OV_DB/Services/TraewellingResyncQueue.cs | 49 ++++++++ OV_DB/Services/TraewellingResyncService.cs | 111 ++++++++++++++++++ OV_DB/Services/TrawellingService.cs | 6 +- OV_DB/Startup.cs | 5 + docs/traewelling-integration.md | 36 ++++-- 19 files changed, 386 insertions(+), 58 deletions(-) create mode 100644 OV_DB.Tests/TraewellingResyncQueueTests.cs create mode 100644 OV_DB/Services/TraewellingResyncQueue.cs create mode 100644 OV_DB/Services/TraewellingResyncService.cs diff --git a/OV_DB.Tests/TraewellingResyncQueueTests.cs b/OV_DB.Tests/TraewellingResyncQueueTests.cs new file mode 100644 index 00000000..eac47b11 --- /dev/null +++ b/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; + +/// +/// 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. +/// +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 FirstAsync(TraewellingResyncQueue queue) + { + await foreach (var userId in queue.DequeueAllAsync(CancellationToken.None)) + return userId; + return 0; + } +} diff --git a/OV_DB.Tests/TrawellingInboxSweepTests.cs b/OV_DB.Tests/TrawellingInboxSweepTests.cs index 82cc3876..20b0a51a 100644 --- a/OV_DB.Tests/TrawellingInboxSweepTests.cs +++ b/OV_DB.Tests/TrawellingInboxSweepTests.cs @@ -35,6 +35,13 @@ private static string Page(string next, params (int Id, string Departure)[] stat return $$$"""{"data":[{{{data}}}],"links":{"next":{{{nextLink}}}}}"""; } + // Progress hands its reports to the thread pool, which would make the order they are + // asserted in a race. + private sealed class InlineProgress(Action report) : IProgress + { + public void Report(T value) => report(value); + } + private sealed class PagedHandler(Dictionary pages) : HttpMessageHandler { public List RequestedPages { get; } = []; @@ -140,14 +147,18 @@ 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(); - var result = await service.SweepInboxAsync(user, TrawellingSweepMode.Full); + var result = await service.SweepInboxAsync(user, TrawellingSweepMode.Full, + new InlineProgress(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] diff --git a/OV_DB/Controllers/TraewellingController.cs b/OV_DB/Controllers/TraewellingController.cs index 62e553ea..32517bd8 100644 --- a/OV_DB/Controllers/TraewellingController.cs +++ b/OV_DB/Controllers/TraewellingController.cs @@ -28,14 +28,17 @@ public class TraewellingController : ControllerBase private readonly OVDBDatabaseContext _dbContext; private readonly ILogger _logger; private readonly ITraewellingRateLimiter _rateLimiter; + private readonly ITraewellingResyncQueue _resyncQueue; public TraewellingController(ITrawellingService trawellingService, - OVDBDatabaseContext dbContext, ILogger logger, ITraewellingRateLimiter rateLimiter) + OVDBDatabaseContext dbContext, ILogger logger, ITraewellingRateLimiter rateLimiter, + ITraewellingResyncQueue resyncQueue) { _trawellingService = trawellingService; _dbContext = dbContext; _logger = logger; _rateLimiter = rateLimiter; + _resyncQueue = resyncQueue; } /// @@ -268,7 +271,8 @@ public async Task GetConnectionStatus() user = userInfo, liveSyncAvailable = _trawellingService.IsLiveSyncAvailable, liveSyncEnabled = user.TraewellingWebhookId.HasValue, - liveSyncHealth = webhookHealth.ToString() + liveSyncHealth = webhookHealth.ToString(), + resyncRunning = _resyncQueue.IsRunning(user.Id) }); } catch (Exception ex) @@ -315,8 +319,10 @@ public async Task GetUnimportedTrips([FromQuery] int page = 1, [F } /// - /// Walk 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. + /// 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 + /// and reports over the hub. /// [HttpPost("resync")] public async Task ResyncFullHistory() @@ -334,24 +340,12 @@ public async Task ResyncFullHistory() if (!_trawellingService.IsConnected(user)) return BadRequest("Träwelling account not connected or tokens expired"); - // Deliberately not the request's token: a long walk that outlives an impatient - // browser or a proxy timeout should still finish and keep what it found. - var result = await _trawellingService.SweepInboxAsync(user, TrawellingSweepMode.Full); - - if (!result.Success && _rateLimiter.IsLimited) - return StatusCode(429, "Träwelling is rate limiting us, please try again in a minute"); - - return Ok(new - { - added = result.Added, - pagesRead = result.PagesRead, - complete = result.Success && result.ReachedEnd - }); + return Accepted(new { queued = _resyncQueue.TryEnqueue(userId.Value) }); } catch (Exception ex) { - _logger.LogError(ex, "Error resyncing Träwelling history"); - return StatusCode(500, "Error resyncing Träwelling history"); + _logger.LogError(ex, "Error queueing Träwelling resync"); + return StatusCode(500, "Error queueing Träwelling resync"); } } diff --git a/OV_DB/Hubs/TraewellingHub.cs b/OV_DB/Hubs/TraewellingHub.cs index 11b13913..fb8892a3 100644 --- a/OV_DB/Hubs/TraewellingHub.cs +++ b/OV_DB/Hubs/TraewellingHub.cs @@ -4,9 +4,10 @@ namespace OV_DB.Hubs; /// -/// 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. /// [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"; } diff --git a/OV_DB/OVDBFrontend/src/app/models/traewelling.model.ts b/OV_DB/OVDBFrontend/src/app/models/traewelling.model.ts index a2f1c246..dc28b9d6 100644 --- a/OV_DB/OVDBFrontend/src/app/models/traewelling.model.ts +++ b/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,17 @@ 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; diff --git a/OV_DB/OVDBFrontend/src/app/traewelling/services/traewelling-live.service.ts b/OV_DB/OVDBFrontend/src/app/traewelling/services/traewelling-live.service.ts index 9935950b..0ccc0320 100644 --- a/OV_DB/OVDBFrontend/src/app/traewelling/services/traewelling-live.service.ts +++ b/OV_DB/OVDBFrontend/src/app/traewelling/services/traewelling-live.service.ts @@ -3,12 +3,18 @@ import { HubConnection, HubConnectionBuilder } from '@microsoft/signalr'; import { Subject } from 'rxjs'; import { environment } from 'src/environments/environment'; import { AuthenticationService } from '../../services/authentication.service'; -import { TrawellingConflict, TrawellingTrip } from '../../models/traewelling.model'; +import { + TrawellingConflict, + TrawellingResyncProgress, + TrawellingResyncResult, + TrawellingTrip +} from '../../models/traewelling.model'; /** * Live updates for the unimported-trips list, pushed by the backend when Träwelling - * webhook events arrive. The hub is JWT-authenticated and only delivers the current - * user's own events. Payloads are JSON strings shaped exactly like the REST responses. + * webhook events arrive and while a full-history resync runs. The hub is JWT-authenticated + * and only delivers the current user's own events. Payloads are JSON strings shaped exactly + * like the REST responses. */ @Injectable({ providedIn: 'root' }) export class TraewellingLiveService { @@ -19,6 +25,8 @@ export class TraewellingLiveService { tripRemoved$ = new Subject(); conflictUpserted$ = new Subject(); conflictRemoved$ = new Subject(); + resyncProgress$ = new Subject(); + resyncFinished$ = new Subject(); connect(): void { if (this.connection) { @@ -52,6 +60,20 @@ export class TraewellingLiveService { connection.on('ConflictRemoved', (statusId: number) => { this.conflictRemoved$.next(statusId); }); + connection.on('ResyncProgress', (progressJson: string) => { + try { + this.resyncProgress$.next(JSON.parse(progressJson) as TrawellingResyncProgress); + } catch (err) { + console.error('Could not parse Träwelling resync progress payload', err); + } + }); + connection.on('ResyncFinished', (resultJson: string) => { + try { + this.resyncFinished$.next(JSON.parse(resultJson) as TrawellingResyncResult); + } catch (err) { + console.error('Could not parse Träwelling resync result payload', err); + } + }); connection .start() diff --git a/OV_DB/OVDBFrontend/src/app/traewelling/services/traewelling.service.ts b/OV_DB/OVDBFrontend/src/app/traewelling/services/traewelling.service.ts index d34796ba..add2d7a1 100644 --- a/OV_DB/OVDBFrontend/src/app/traewelling/services/traewelling.service.ts +++ b/OV_DB/OVDBFrontend/src/app/traewelling/services/traewelling.service.ts @@ -19,7 +19,7 @@ import { TraewellingAlert, TrawellingConflict, TrawellingConflictAction, - TrawellingResyncResult + TrawellingResyncQueued } from '../../models/traewelling.model'; import { TraewellingTagMapping } from '../../models/user-profile.model'; import { ApiService } from '../../services/api.service'; @@ -59,8 +59,9 @@ export class TrawellingService { ); } - async resyncFullHistory(): Promise { - return firstValueFrom(this.http.post(`${this.baseUrl}/resync`, {}) + /** Queues the walk; it reports over the hub, not in this response. */ + async resyncFullHistory(): Promise { + return firstValueFrom(this.http.post(`${this.baseUrl}/resync`, {}) .pipe(first()) ); } diff --git a/OV_DB/OVDBFrontend/src/app/traewelling/traewelling.component.html b/OV_DB/OVDBFrontend/src/app/traewelling/traewelling.component.html index 23222360..602682c7 100644 --- a/OV_DB/OVDBFrontend/src/app/traewelling/traewelling.component.html +++ b/OV_DB/OVDBFrontend/src/app/traewelling/traewelling.component.html @@ -24,7 +24,17 @@ } - + + @if (isResyncing) { +
+ + + {{ (resyncProgress ? 'TRAEWELLING.RESYNC_PROGRESS' : 'TRAEWELLING.RESYNC_RUNNING') | translate:{ + pages: resyncProgress?.pagesRead, count: resyncProgress?.added } }} + +
+ } + diff --git a/OV_DB/OVDBFrontend/src/app/traewelling/traewelling.component.scss b/OV_DB/OVDBFrontend/src/app/traewelling/traewelling.component.scss index 447fbdcb..236d0f92 100644 --- a/OV_DB/OVDBFrontend/src/app/traewelling/traewelling.component.scss +++ b/OV_DB/OVDBFrontend/src/app/traewelling/traewelling.component.scss @@ -34,6 +34,12 @@ align-self: center; } + .resync-progress { + display: flex; + align-items: center; + gap: 8px; + } + .conflicts-container { margin-bottom: 16px; diff --git a/OV_DB/OVDBFrontend/src/app/traewelling/traewelling.component.ts b/OV_DB/OVDBFrontend/src/app/traewelling/traewelling.component.ts index 118db627..4a7607bc 100644 --- a/OV_DB/OVDBFrontend/src/app/traewelling/traewelling.component.ts +++ b/OV_DB/OVDBFrontend/src/app/traewelling/traewelling.component.ts @@ -18,7 +18,9 @@ import { TrawellingConflict, TrawellingConflictAction, TraewellingAlert, - TraewellingAlertTranslation + TraewellingAlertTranslation, + TrawellingResyncProgress, + TrawellingResyncResult } from '../models/traewelling.model'; import { TripCardComponent } from './components/trip-card/trip-card.component'; import { ConflictCardComponent } from './components/conflict-card/conflict-card.component'; @@ -55,6 +57,7 @@ export class TrawellingComponent implements OnInit, OnDestroy { alerts: TraewellingAlert[] = []; isLoading = true; isResyncing = false; + resyncProgress: TrawellingResyncProgress | null = null; isLoadingMore = false; hasMorePages = false; currentPage = 1; @@ -65,9 +68,9 @@ export class TrawellingComponent implements OnInit, OnDestroy { await this.loadConnectionStatus(); if (this.connectionStatus?.connected) { await Promise.all([this.loadTrips(), this.loadAlerts(), this.loadConflicts()]); - if (this.connectionStatus.liveSyncEnabled) { - this.startLiveUpdates(); - } + // Also without live sync: a resync started here reports over the same hub. + this.startLiveUpdates(); + this.isResyncing = this.connectionStatus.resyncRunning === true; } this.isLoading = false; } @@ -89,6 +92,8 @@ export class TrawellingComponent implements OnInit, OnDestroy { this.liveService.tripRemoved$.subscribe(statusId => this.removeTrip(statusId)), this.liveService.conflictUpserted$.subscribe(conflict => this.upsertConflict(conflict)), this.liveService.conflictRemoved$.subscribe(statusId => this.removeConflict(statusId)), + this.liveService.resyncProgress$.subscribe(progress => this.resyncProgress = progress), + this.liveService.resyncFinished$.subscribe(result => this.finishResync(result)), ); } @@ -203,25 +208,38 @@ export class TrawellingComponent implements OnInit, OnDestroy { } } + /** + * Only queues the walk — it takes minutes and reports over the hub, so the spinner runs + * until a ResyncFinished arrives rather than until this request returns. + */ async resyncFullHistory() { if (this.isResyncing) return; this.isResyncing = true; + this.resyncProgress = null; try { - const result = await this.trawellingService.resyncFullHistory(); - const messageKey = !result.complete - ? 'TRAEWELLING.RESYNC_INCOMPLETE' - : result.added > 0 ? 'TRAEWELLING.RESYNC_FOUND' : 'TRAEWELLING.RESYNC_NOTHING_FOUND'; + await this.trawellingService.resyncFullHistory(); this.snackBar.open( - this.translateService.instant(messageKey, { count: result.added }), + this.translateService.instant('TRAEWELLING.RESYNC_STARTED'), this.translateService.instant('CLOSE'), { duration: 5000 }); - if (result.added > 0) { - await this.loadTrips(); - } } catch { - // The global error interceptor already showed a toast - } finally { this.isResyncing = false; + // The global error interceptor already showed a toast + } + } + + private async finishResync(result: TrawellingResyncResult) { + this.isResyncing = false; + this.resyncProgress = null; + const messageKey = !result.complete + ? 'TRAEWELLING.RESYNC_INCOMPLETE' + : result.added > 0 ? 'TRAEWELLING.RESYNC_FOUND' : 'TRAEWELLING.RESYNC_NOTHING_FOUND'; + this.snackBar.open( + this.translateService.instant(messageKey, { count: result.added }), + this.translateService.instant('CLOSE'), + { duration: 8000 }); + if (result.added > 0) { + await this.loadTrips(); } } diff --git a/OV_DB/OVDBFrontend/src/assets/i18n/en.json b/OV_DB/OVDBFrontend/src/assets/i18n/en.json index c2c0b087..487783f9 100644 --- a/OV_DB/OVDBFrontend/src/assets/i18n/en.json +++ b/OV_DB/OVDBFrontend/src/assets/i18n/en.json @@ -729,6 +729,9 @@ "CATEGORY_UNKNOWN": "Unknown", "REFRESH": "Check Träwelling for new trips", "RESYNC": "Search your whole Träwelling history for trips that were skipped", + "RESYNC_STARTED": "Searching your whole Träwelling history, this takes a few minutes. You can keep using the site.", + "RESYNC_RUNNING": "Searching your whole Träwelling history...", + "RESYNC_PROGRESS": "Searching your whole Träwelling history: {{pages}} pages read, {{count}} trips found", "RESYNC_FOUND": "Skipped trips found: {{count}}", "RESYNC_NOTHING_FOUND": "No skipped trips found, your whole Träwelling history is here", "RESYNC_INCOMPLETE": "Could not read your whole history, try again. Trips found so far: {{count}}", diff --git a/OV_DB/OVDBFrontend/src/assets/i18n/nl.json b/OV_DB/OVDBFrontend/src/assets/i18n/nl.json index 9cbdb777..d94eea69 100644 --- a/OV_DB/OVDBFrontend/src/assets/i18n/nl.json +++ b/OV_DB/OVDBFrontend/src/assets/i18n/nl.json @@ -727,6 +727,9 @@ "CATEGORY_UNKNOWN": "Onbekend", "REFRESH": "Träwelling controleren op nieuwe reizen", "RESYNC": "Je hele Träwelling-geschiedenis doorzoeken op overgeslagen reizen", + "RESYNC_STARTED": "Je hele Träwelling-geschiedenis wordt doorzocht, dit duurt een paar minuten. Je kunt de site gewoon blijven gebruiken.", + "RESYNC_RUNNING": "Je hele Träwelling-geschiedenis wordt doorzocht...", + "RESYNC_PROGRESS": "Je hele Träwelling-geschiedenis wordt doorzocht: {{pages}} pagina's gelezen, {{count}} reizen gevonden", "RESYNC_FOUND": "Overgeslagen reizen gevonden: {{count}}", "RESYNC_NOTHING_FOUND": "Geen overgeslagen reizen gevonden, je hele Träwelling-geschiedenis staat er al in", "RESYNC_INCOMPLETE": "Kon je hele geschiedenis niet lezen, probeer het opnieuw. Tot nu toe gevonden: {{count}}", diff --git a/OV_DB/Services/ITrawellingService.cs b/OV_DB/Services/ITrawellingService.cs index 786d6db7..d585e721 100644 --- a/OV_DB/Services/ITrawellingService.cs +++ b/OV_DB/Services/ITrawellingService.cs @@ -31,6 +31,9 @@ public enum TrawellingSweepMode /// True when the walk ran out of pages rather than stopping at a cap. public record TrawellingSweepResult(bool Success, int Added, int PagesRead, bool ReachedEnd); + /// How far a sweep has got, reported after every page. + public record TrawellingSweepProgress(int PagesRead, int Added); + public interface ITrawellingService { /// @@ -103,8 +106,9 @@ public interface ITrawellingService /// /// User to sweep for /// How far back to read the listing + /// Reported after every page; the only sign of life during a full walk /// Cancellation token - Task SweepInboxAsync(User user, TrawellingSweepMode mode = TrawellingSweepMode.WhenStale, System.Threading.CancellationToken cancellationToken = default); + Task SweepInboxAsync(User user, TrawellingSweepMode mode = TrawellingSweepMode.WhenStale, IProgress progress = null, System.Threading.CancellationToken cancellationToken = default); /// /// Remove a status from the inbox after it has been imported or ignored diff --git a/OV_DB/Services/TraewellingInboxSweepService.cs b/OV_DB/Services/TraewellingInboxSweepService.cs index 9aa2cbbc..8e2630fc 100644 --- a/OV_DB/Services/TraewellingInboxSweepService.cs +++ b/OV_DB/Services/TraewellingInboxSweepService.cs @@ -55,7 +55,7 @@ private async Task SweepAllUsersAsync(CancellationToken cancellationToken) foreach (var user in connectedUsers) { cancellationToken.ThrowIfCancellationRequested(); - if ((await trawellingService.SweepInboxAsync(user, TrawellingSweepMode.Force, cancellationToken)).Success) + if ((await trawellingService.SweepInboxAsync(user, TrawellingSweepMode.Force, cancellationToken: cancellationToken)).Success) swept++; // Daily job, so latency is free — don't burst through the shared rate limit await Task.Delay(TimeSpan.FromSeconds(5), cancellationToken); diff --git a/OV_DB/Services/TraewellingResyncQueue.cs b/OV_DB/Services/TraewellingResyncQueue.cs new file mode 100644 index 00000000..193baef9 --- /dev/null +++ b/OV_DB/Services/TraewellingResyncQueue.cs @@ -0,0 +1,49 @@ +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Channels; + +namespace OV_DB.Services +{ + /// + /// The users waiting for a full-history resync. One at a time per user: the walk takes + /// minutes, and a second one started alongside it would re-read the same pages and race + /// the first on the inbox's unique index. + /// + public interface ITraewellingResyncQueue + { + /// False when this user already has a resync queued or running. + bool TryEnqueue(int userId); + + bool IsRunning(int userId); + + IAsyncEnumerable DequeueAllAsync(CancellationToken cancellationToken); + + void MarkFinished(int userId); + } + + public class TraewellingResyncQueue : ITraewellingResyncQueue + { + private readonly Channel _queue = Channel.CreateUnbounded(); + private readonly ConcurrentDictionary _pending = new(); + + public bool TryEnqueue(int userId) + { + if (!_pending.TryAdd(userId, 0)) + return false; + + if (_queue.Writer.TryWrite(userId)) + return true; + + _pending.TryRemove(userId, out _); + return false; + } + + public bool IsRunning(int userId) => _pending.ContainsKey(userId); + + public IAsyncEnumerable DequeueAllAsync(CancellationToken cancellationToken) => + _queue.Reader.ReadAllAsync(cancellationToken); + + public void MarkFinished(int userId) => _pending.TryRemove(userId, out _); + } +} diff --git a/OV_DB/Services/TraewellingResyncService.cs b/OV_DB/Services/TraewellingResyncService.cs new file mode 100644 index 00000000..3f356293 --- /dev/null +++ b/OV_DB/Services/TraewellingResyncService.cs @@ -0,0 +1,111 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.AspNetCore.SignalR; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using Newtonsoft.Json; +using OV_DB.Hubs; +using OVDB_database.Database; + +namespace OV_DB.Services +{ + /// + /// Runs the full-history resyncs the Träwelling page asks for. Walking a whole history + /// takes minutes — longer than any request survives — so the endpoint only queues, and + /// the page hears about it over the hub. + /// + public class TraewellingResyncService( + IServiceProvider serviceProvider, + ITraewellingResyncQueue queue, + IHubContext hubContext, + ILogger logger) : BackgroundService + { + // A page at a time is too chatty for a thousand-page walk and says nothing new. + private const int ProgressEveryPages = 5; + + /// + /// Not : that posts each report to the thread pool, which + /// reorders them and races the counter this throttles on. + /// + private sealed class InlineProgress(Action report) : IProgress + { + public void Report(T value) => report(value); + } + + protected override async Task ExecuteAsync(CancellationToken stoppingToken) + { + await foreach (var userId in queue.DequeueAllAsync(stoppingToken)) + { + try + { + await ResyncAsync(userId, stoppingToken); + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + logger.LogError(ex, "Error resyncing Träwelling history for user {UserId}", userId); + await PublishAsync(userId, TraewellingHub.ResyncFinishedMethod, + new { added = 0, pagesRead = 0, complete = false }); + } + finally + { + queue.MarkFinished(userId); + } + } + } + + private async Task ResyncAsync(int userId, CancellationToken cancellationToken) + { + using var scope = serviceProvider.CreateScope(); + var dbContext = scope.ServiceProvider.GetRequiredService(); + var trawellingService = scope.ServiceProvider.GetRequiredService(); + + var user = await dbContext.Users.FirstOrDefaultAsync(u => u.Id == userId, cancellationToken); + if (user == null || !trawellingService.IsConnected(user)) + { + await PublishAsync(userId, TraewellingHub.ResyncFinishedMethod, + new { added = 0, pagesRead = 0, complete = false }); + return; + } + + var lastReported = 0; + var progress = new InlineProgress(p => + { + if (p.PagesRead - lastReported < ProgressEveryPages) + return; + lastReported = p.PagesRead; + // Not awaited: the walk should not wait on a hub delivery, and a failed one + // is logged and forgotten — the finish event carries the totals anyway. + _ = PublishAsync(userId, TraewellingHub.ResyncProgressMethod, + new { pagesRead = p.PagesRead, added = p.Added }); + }); + + var result = await trawellingService.SweepInboxAsync(user, TrawellingSweepMode.Full, progress, cancellationToken); + + logger.LogInformation("Träwelling resync for user {UserId}: {Added} added over {Pages} pages, complete {Complete}", + userId, result.Added, result.PagesRead, result.Success && result.ReachedEnd); + + await PublishAsync(userId, TraewellingHub.ResyncFinishedMethod, new + { + added = result.Added, + pagesRead = result.PagesRead, + complete = result.Success && result.ReachedEnd + }); + } + + private async Task PublishAsync(int userId, string method, object payload) + { + try + { + await hubContext.Clients.User(userId.ToString()) + .SendAsync(method, JsonConvert.SerializeObject(payload)); + } + catch (Exception ex) + { + logger.LogError(ex, "Error publishing Träwelling resync {Method} for user {UserId}", method, userId); + } + } + } +} diff --git a/OV_DB/Services/TrawellingService.cs b/OV_DB/Services/TrawellingService.cs index 78db5c6b..35b134c9 100644 --- a/OV_DB/Services/TrawellingService.cs +++ b/OV_DB/Services/TrawellingService.cs @@ -329,7 +329,7 @@ public async Task GetUserInfoAsync(User user) private static readonly TimeSpan SweepStaleness = TimeSpan.FromHours(1); - public async Task SweepInboxAsync(User user, TrawellingSweepMode mode = TrawellingSweepMode.WhenStale, CancellationToken cancellationToken = default) + public async Task SweepInboxAsync(User user, TrawellingSweepMode mode = TrawellingSweepMode.WhenStale, IProgress progress = null, CancellationToken cancellationToken = default) { var added = 0; var pagesRead = 0; @@ -433,6 +433,8 @@ public async Task SweepInboxAsync(User user, TrawellingSw if (newOnPage > 0) await _dbContext.SaveChangesAsync(cancellationToken); + progress?.Report(new TrawellingSweepProgress(pagesRead, added)); + if (string.IsNullOrEmpty(statusesResponse.Links?.Next)) { reachedEnd = true; @@ -499,7 +501,7 @@ public async Task GetOptimizedTripsAsync(User user, int // The inbox is the source of the list; the statuses API is only touched by the // sweep (skipped when fresh, forced by the frontend's refresh action). A failed // sweep still returns the current — possibly stale — inbox contents. - await SweepInboxAsync(user, refresh ? TrawellingSweepMode.Force : TrawellingSweepMode.WhenStale, cancellationToken); + await SweepInboxAsync(user, refresh ? TrawellingSweepMode.Force : TrawellingSweepMode.WhenStale, cancellationToken: cancellationToken); const int pageSize = 15; var query = _dbContext.TrawellingInboxStatuses diff --git a/OV_DB/Startup.cs b/OV_DB/Startup.cs index 4db83db3..5006f682 100644 --- a/OV_DB/Startup.cs +++ b/OV_DB/Startup.cs @@ -193,6 +193,11 @@ public void ConfigureServices(IServiceCollection services) services.AddHostedService(); } + // Not behind that flag: this one only acts on a user pressing the button, and + // then it has to run wherever they pressed it. + services.AddSingleton(); + services.AddHostedService(); + // Register named HttpClients for different services to avoid socket exhaustion services.AddHttpClient("OSM", client => { diff --git a/docs/traewelling-integration.md b/docs/traewelling-integration.md index a2f86037..3bddabfe 100644 --- a/docs/traewelling-integration.md +++ b/docs/traewelling-integration.md @@ -106,20 +106,20 @@ Imports a specific Träwelling trip as an OVDB RouteInstance. ``` ### POST /api/traewelling/resync -Walks the whole statuses listing, past pages holding nothing new, and adds everything still -unknown to the inbox. The history button next to refresh on the Träwelling page. +Queues a walk of the whole statuses listing, past pages holding nothing new, adding everything +still unknown to the inbox. The history button next to refresh on the Träwelling page. + +A whole history takes minutes to walk — longer than a request survives behind the reverse +proxy — so this returns **202 Accepted** immediately and `TraewellingResyncService` does the +walking. One walk per user: `queued: false` means one was already running. -**Response:** ```json -{ - "added": 3, - "pagesRead": 27, - "complete": true -} +{ "queued": true } ``` -`complete: false` means the walk stopped on an error or the 1000-page safety limit, so older -check-ins may still be missing; a rerun starts at page 1 again. +Progress and outcome arrive over the hub (see Live updates), not in this response. +`resyncRunning` on `GET /status` says whether one is still going, so a page opened midway +shows it too. ### GET /api/traewelling/stats Returns statistics about the user's Träwelling integration. @@ -137,6 +137,22 @@ Returns statistics about the user's Träwelling integration. ### DELETE /api/traewelling/disconnect Disconnects the Träwelling account by removing stored tokens. +## Live updates (`/traewellingHub`) + +JWT-authenticated SignalR hub, delivering only the connected user's own events. Every payload +is a JSON string in the same shape as the REST responses, so the client parses rather than +maps it. + +| Event | Sent when | +| --- | --- | +| `PendingTripUpserted` / `PendingTripRemoved` | A webhook event changed the unimported list | +| `ConflictUpserted` / `ConflictRemoved` | A webhook event changed the conflict list | +| `ResyncProgress` | Every five pages of a running resync: `{ pagesRead, added }` | +| `ResyncFinished` | A resync ended: `{ added, pagesRead, complete }` | + +`complete: false` means the walk stopped on an error or the 1000-page safety limit, so older +check-ins may still be missing; a rerun starts at page 1 again. + ## Authentication & Authorization - All endpoints except `/callback` require JWT authentication