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 new file mode 100644 index 00000000..20b0a51a --- /dev/null +++ b/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; + +/// +/// 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}}}}}"""; + } + + // 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; } = []; + + 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 reported = new List(); + + 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] + 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..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) @@ -314,6 +318,37 @@ public async Task GetUnimportedTrips([FromQuery] int page = 1, [F } } + /// + /// 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() + { + 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"); + } + } + /// /// Ignore a Träwelling status so it doesn't show up in unimported list /// 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 8a3e472d..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,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[]; 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 fbb19a3c..add2d7a1 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, + TrawellingResyncQueued } from '../../models/traewelling.model'; import { TraewellingTagMapping } from '../../models/user-profile.model'; import { ApiService } from '../../services/api.service'; @@ -58,6 +59,13 @@ export class TrawellingService { ); } + /** 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()) + ); + } + 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..602682c7 100644 --- a/OV_DB/OVDBFrontend/src/app/traewelling/traewelling.component.html +++ b/OV_DB/OVDBFrontend/src/app/traewelling/traewelling.component.html @@ -18,9 +18,23 @@ [attr.aria-label]="'TRAEWELLING.REFRESH' | translate"> refresh + } - + + @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 e4b4adbc..236d0f92 100644 --- a/OV_DB/OVDBFrontend/src/app/traewelling/traewelling.component.scss +++ b/OV_DB/OVDBFrontend/src/app/traewelling/traewelling.component.scss @@ -30,6 +30,16 @@ align-self: center; } + .resync-button { + 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 71acb8be..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'; @@ -54,6 +56,8 @@ export class TrawellingComponent implements OnInit, OnDestroy { conflicts: TrawellingConflict[] = []; alerts: TraewellingAlert[] = []; isLoading = true; + isResyncing = false; + resyncProgress: TrawellingResyncProgress | null = null; isLoadingMore = false; hasMorePages = false; currentPage = 1; @@ -64,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; } @@ -88,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)), ); } @@ -202,6 +208,41 @@ 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 { + await this.trawellingService.resyncFullHistory(); + this.snackBar.open( + this.translateService.instant('TRAEWELLING.RESYNC_STARTED'), + this.translateService.instant('CLOSE'), + { duration: 5000 }); + } catch { + 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(); + } + } + 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..487783f9 100644 --- a/OV_DB/OVDBFrontend/src/assets/i18n/en.json +++ b/OV_DB/OVDBFrontend/src/assets/i18n/en.json @@ -728,6 +728,13 @@ "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_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}}", "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..d94eea69 100644 --- a/OV_DB/OVDBFrontend/src/assets/i18n/nl.json +++ b/OV_DB/OVDBFrontend/src/assets/i18n/nl.json @@ -726,6 +726,13 @@ "CATEGORY_TAXI": "Taxi", "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}}", "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..d585e721 100644 --- a/OV_DB/Services/ITrawellingService.cs +++ b/OV_DB/Services/ITrawellingService.cs @@ -7,6 +7,33 @@ 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); + + /// How far a sweep has got, reported after every page. + public record TrawellingSweepProgress(int PagesRead, int Added); + public interface ITrawellingService { /// @@ -75,14 +102,13 @@ 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 + /// Reported after every page; the only sign of life during a full walk /// 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, 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 133ce0bc..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, force: true, cancellationToken)) + 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 cabcbc89..35b134c9 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, IProgress progress = null, 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,28 @@ 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); + + progress?.Report(new TrawellingSweepProgress(pagesRead, added)); + + 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 +473,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 +482,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 +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, force: refresh, 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 00127318..3bddabfe 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 +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. + +```json +{ "queued": true } +``` + +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. @@ -118,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