diff --git a/Jobs/RssFeedJob.cs b/Jobs/RssFeedJob.cs index b10a9ec..7978671 100644 --- a/Jobs/RssFeedJob.cs +++ b/Jobs/RssFeedJob.cs @@ -17,9 +17,10 @@ public class RssFeedJob(DB db, RssFeedService rssFeed, DiscordWebhookService dis { public async Task Execute(IJobExecutionContext context) { + CancellationToken cancellationToken = context.CancellationToken; List subscriptions = await db.RssSubscriptions .Include(s => s.Webhook) - .ToListAsync(); + .ToListAsync(cancellationToken); if (subscriptions.Count == 0) return; @@ -28,10 +29,11 @@ public async Task Execute(IJobExecutionContext context) foreach (IGrouping group in subscriptions.GroupBy(s => s.FeedUrl)) { + cancellationToken.ThrowIfCancellationRequested(); string feedUrl = group.Key; List subs = group.ToList(); - (string? _, string? _, IReadOnlyList entries) = await rssFeed.FetchAsync(feedUrl); + (string? _, string? _, IReadOnlyList entries) = await rssFeed.FetchAsync(feedUrl, cancellationToken); if (entries.Count == 0) continue; @@ -39,18 +41,18 @@ public async Task Execute(IJobExecutionContext context) HashSet seen = (await db.RssSeenEntries .Where(v => v.FeedUrl == feedUrl && entryIds.Contains(v.EntryId)) .Select(v => v.EntryId) - .ToListAsync()) + .ToListAsync(cancellationToken)) .ToHashSet(); // If nothing from this feed has ever been seen, this is an initial run: mark // everything seen and only post the latest entry to avoid backfilling history. // Check all history for the feed because older seen entries may have rolled out of // the feed's current response. - bool initialSeed = !await HasFeedHistoryAsync(db, feedUrl); + bool initialSeed = !await HasFeedHistoryAsync(db, feedUrl, cancellationToken); if (initialSeed) { RssFeedService.FeedEntry latest = entries.OrderByDescending(e => e.Published).First(); - if (!await DispatchAsync(latest, subs, SendAsync)) + if (!await DispatchAsync(latest, subs, (sub, content) => SendAsync(sub, content, cancellationToken), cancellationToken)) continue; foreach (RssFeedService.FeedEntry entry in entries) @@ -64,10 +66,11 @@ public async Task Execute(IJobExecutionContext context) foreach (RssFeedService.FeedEntry entry in entries.OrderBy(e => e.Published)) { + cancellationToken.ThrowIfCancellationRequested(); if (seen.Contains(entry.EntryId)) continue; - if (!await DispatchAsync(entry, subs, SendAsync)) + if (!await DispatchAsync(entry, subs, (sub, content) => SendAsync(sub, content, cancellationToken), cancellationToken)) continue; db.RssSeenEntries.Add(new RssSeenEntry { FeedUrl = feedUrl, EntryId = entry.EntryId, SeenAt = DateTime.UtcNow }); @@ -77,13 +80,14 @@ public async Task Execute(IJobExecutionContext context) } if (changed) - await db.SaveChangesAsync(); + await db.SaveChangesAsync(cancellationToken); } internal static async Task DispatchAsync( RssFeedService.FeedEntry entry, IReadOnlyList subs, - Func> sendAsync) + Func> sendAsync, + CancellationToken cancellationToken = default) { string content = !string.IsNullOrWhiteSpace(entry.Link) ? entry.Link : entry.Title; if (string.IsNullOrWhiteSpace(content)) @@ -92,6 +96,7 @@ internal static async Task DispatchAsync( bool allSucceeded = true; foreach (RssSubscription sub in subs) { + cancellationToken.ThrowIfCancellationRequested(); if (!await sendAsync(sub, content)) allSucceeded = false; } @@ -99,10 +104,10 @@ internal static async Task DispatchAsync( return allSucceeded; } - internal static Task HasFeedHistoryAsync(DB db, string feedUrl) => - db.RssSeenEntries.AnyAsync(entry => entry.FeedUrl == feedUrl); + internal static Task HasFeedHistoryAsync(DB db, string feedUrl, CancellationToken cancellationToken = default) => + db.RssSeenEntries.AnyAsync(entry => entry.FeedUrl == feedUrl, cancellationToken); - private async Task SendAsync(RssSubscription sub, string content) + private async Task SendAsync(RssSubscription sub, string content, CancellationToken cancellationToken) { if (sub.Webhook == null) { @@ -110,7 +115,7 @@ private async Task SendAsync(RssSubscription sub, string content) return false; } - bool ok = await discordWebhook.SendAsync(sub.Webhook.WebhookId, sub.Webhook.Token, content, sub.DisplayName, sub.AvatarUrl); + bool ok = await discordWebhook.SendAsync(sub.Webhook.WebhookId, sub.Webhook.Token, content, sub.DisplayName, sub.AvatarUrl, cancellationToken); if (!ok) logsService.Log($"RssFeedJob: failed to post entry from {sub.FeedUrl} to channel {sub.ChannelDiscordId}", LogSeverity.Warning); diff --git a/Jobs/YoutubeRssJob.cs b/Jobs/YoutubeRssJob.cs index 7e9929b..0daf006 100644 --- a/Jobs/YoutubeRssJob.cs +++ b/Jobs/YoutubeRssJob.cs @@ -20,9 +20,10 @@ public class YoutubeRssJob(DB db, YoutubeFeedService youtubeFeed, DiscordWebhook public async Task Execute(IJobExecutionContext context) { + CancellationToken cancellationToken = context.CancellationToken; List subscriptions = await db.YoutubeSubscriptions .Include(s => s.Webhook) - .ToListAsync(); + .ToListAsync(cancellationToken); if (subscriptions.Count == 0) return; @@ -31,10 +32,11 @@ public async Task Execute(IJobExecutionContext context) foreach (IGrouping group in subscriptions.GroupBy(s => s.YoutubeChannelId)) { + cancellationToken.ThrowIfCancellationRequested(); string youtubeChannelId = group.Key; List subs = group.ToList(); - (string? channelTitle, IReadOnlyList entries) = await youtubeFeed.FetchFeedAsync(youtubeChannelId); + (string? channelTitle, IReadOnlyList entries) = await youtubeFeed.FetchFeedAsync(youtubeChannelId, cancellationToken); if (entries.Count == 0) continue; @@ -42,7 +44,7 @@ public async Task Execute(IJobExecutionContext context) string username = !string.IsNullOrWhiteSpace(channelTitle) ? channelTitle! : subs[0].YoutubeChannelTitle; string? avatar = subs.Select(s => s.YoutubeAvatarUrl).FirstOrDefault(a => !string.IsNullOrWhiteSpace(a)); if (string.IsNullOrWhiteSpace(avatar)) - avatar = await YoutubeUtils.GetChannelAvatarAsync(HttpClient, youtubeChannelId); + avatar = await YoutubeUtils.GetChannelAvatarAsync(HttpClient, youtubeChannelId, cancellationToken); foreach (YoutubeSubscription sub in subs) { @@ -62,18 +64,18 @@ public async Task Execute(IJobExecutionContext context) HashSet seen = (await db.YoutubeSeenVideos .Where(v => videoIds.Contains(v.VideoId)) .Select(v => v.VideoId) - .ToListAsync()) + .ToListAsync(cancellationToken)) .ToHashSet(); // If nothing from this channel has ever been seen, this is an initial run for it: // mark everything seen and only post the latest video to avoid backfilling history. // Check all history for the channel because older seen videos may have rolled out of // the feed's current response. - bool initialSeed = !await HasFeedHistoryAsync(db, youtubeChannelId); + bool initialSeed = !await HasFeedHistoryAsync(db, youtubeChannelId, cancellationToken); if (initialSeed) { YoutubeFeedService.VideoEntry latest = entries.OrderByDescending(e => e.Published).First(); - if (!await DispatchAsync(subs, sub => SendAsync(sub, latest, username, avatar))) + if (!await DispatchAsync(subs, sub => SendAsync(sub, latest, username, avatar, cancellationToken), cancellationToken)) continue; foreach (YoutubeFeedService.VideoEntry entry in entries) @@ -87,10 +89,11 @@ public async Task Execute(IJobExecutionContext context) foreach (YoutubeFeedService.VideoEntry entry in entries.OrderBy(e => e.Published)) { + cancellationToken.ThrowIfCancellationRequested(); if (seen.Contains(entry.VideoId)) continue; - if (!await DispatchAsync(subs, sub => SendAsync(sub, entry, username, avatar))) + if (!await DispatchAsync(subs, sub => SendAsync(sub, entry, username, avatar, cancellationToken), cancellationToken)) continue; db.YoutubeSeenVideos.Add(new YoutubeSeenVideo { YoutubeChannelId = youtubeChannelId, VideoId = entry.VideoId, SeenAt = DateTime.UtcNow }); @@ -100,16 +103,18 @@ public async Task Execute(IJobExecutionContext context) } if (changed) - await db.SaveChangesAsync(); + await db.SaveChangesAsync(cancellationToken); } internal static async Task DispatchAsync( IReadOnlyList subs, - Func> sendAsync) + Func> sendAsync, + CancellationToken cancellationToken = default) { bool allSucceeded = true; foreach (YoutubeSubscription sub in subs) { + cancellationToken.ThrowIfCancellationRequested(); if (!await sendAsync(sub)) allSucceeded = false; } @@ -117,14 +122,15 @@ internal static async Task DispatchAsync( return allSucceeded; } - internal static Task HasFeedHistoryAsync(DB db, string youtubeChannelId) => - db.YoutubeSeenVideos.AnyAsync(video => video.YoutubeChannelId == youtubeChannelId); + internal static Task HasFeedHistoryAsync(DB db, string youtubeChannelId, CancellationToken cancellationToken = default) => + db.YoutubeSeenVideos.AnyAsync(video => video.YoutubeChannelId == youtubeChannelId, cancellationToken); private async Task SendAsync( YoutubeSubscription sub, YoutubeFeedService.VideoEntry entry, string username, - string? avatar) + string? avatar, + CancellationToken cancellationToken) { if (sub.Webhook == null) { @@ -132,7 +138,7 @@ private async Task SendAsync( return false; } - bool ok = await discordWebhook.SendAsync(sub.Webhook.WebhookId, sub.Webhook.Token, entry.Link, username, avatar); + bool ok = await discordWebhook.SendAsync(sub.Webhook.WebhookId, sub.Webhook.Token, entry.Link, username, avatar, cancellationToken); if (!ok) logsService.Log($"YoutubeRssJob: failed to post {entry.VideoId} to channel {sub.ChannelDiscordId}", LogSeverity.Warning); diff --git a/Morpheus.Tests/RssFeedJobTests.cs b/Morpheus.Tests/RssFeedJobTests.cs index afb528b..bc2dee3 100644 --- a/Morpheus.Tests/RssFeedJobTests.cs +++ b/Morpheus.Tests/RssFeedJobTests.cs @@ -98,4 +98,26 @@ [new RssSubscription()], Assert.True(succeeded); Assert.False(sent); } + + [Fact] + public async Task DispatchAsync_WhenCallerCancels_PropagatesCancellationBeforeDelivery() + { + RssFeedService.FeedEntry entry = new("entry-1", "Entry", "https://example.com/entry-1", DateTime.UtcNow); + using CancellationTokenSource cts = new(); + await cts.CancelAsync(); + bool sent = false; + + await Assert.ThrowsAnyAsync(() => + RssFeedJob.DispatchAsync( + entry, + [new RssSubscription()], + (_, _) => + { + sent = true; + return Task.FromResult(true); + }, + cts.Token)); + + Assert.False(sent); + } } \ No newline at end of file diff --git a/Morpheus.Tests/YoutubeRssJobTests.cs b/Morpheus.Tests/YoutubeRssJobTests.cs index 24fa5aa..2ead97e 100644 --- a/Morpheus.Tests/YoutubeRssJobTests.cs +++ b/Morpheus.Tests/YoutubeRssJobTests.cs @@ -79,4 +79,24 @@ Task SendAsync(YoutubeSubscription _) Assert.True(secondSucceeded); Assert.Equal(4, attempts); } + + [Fact] + public async Task DispatchAsync_WhenCallerCancels_PropagatesCancellationBeforeDelivery() + { + using CancellationTokenSource cts = new(); + await cts.CancelAsync(); + bool sent = false; + + await Assert.ThrowsAnyAsync(() => + YoutubeRssJob.DispatchAsync( + [new YoutubeSubscription()], + _ => + { + sent = true; + return Task.FromResult(true); + }, + cts.Token)); + + Assert.False(sent); + } } diff --git a/Morpheus.Tests/YoutubeUtilsTests.cs b/Morpheus.Tests/YoutubeUtilsTests.cs index 741975b..6c0b132 100644 --- a/Morpheus.Tests/YoutubeUtilsTests.cs +++ b/Morpheus.Tests/YoutubeUtilsTests.cs @@ -27,6 +27,18 @@ public async Task ResolveChannelIdAsync_DoesNotRequestNonYoutubeUrls(string inpu Assert.Empty(handler.RequestedUris); } + [Fact] + public async Task GetChannelAvatarAsync_WhenCallerCancels_PropagatesCancellation() + { + RecordingHandler handler = new(_ => throw new InvalidOperationException("Unexpected HTTP request.")); + using HttpClient httpClient = new(handler); + using CancellationTokenSource cts = new(); + await cts.CancelAsync(); + + await Assert.ThrowsAnyAsync(() => + YoutubeUtils.GetChannelAvatarAsync(httpClient, "UC123", cts.Token)); + } + [Theory] [InlineData("https://www.youtube.com/@channel", "/@channel", "")] [InlineData("youtube.com/user/channel", "/user/channel", "")] @@ -74,6 +86,9 @@ private sealed class RecordingHandler(Func SendAsync(HttpRequestMessage request, CancellationToken cancellationToken) { + if (cancellationToken.IsCancellationRequested) + return Task.FromCanceled(cancellationToken); + if (request.RequestUri != null) RequestedUris.Add(request.RequestUri); diff --git a/Utilities/YoutubeUtils.cs b/Utilities/YoutubeUtils.cs index 350d7fb..9da9297 100644 --- a/Utilities/YoutubeUtils.cs +++ b/Utilities/YoutubeUtils.cs @@ -154,7 +154,7 @@ private static bool IsSupportedYoutubePath(string path) => /// Uses the Innertube (youtubei) browse API to fetch a channel's avatar URL. /// Returns null on error or if thumbnails are not found. /// - public static async Task GetChannelAvatarAsync(HttpClient httpClient, string channelId) + public static async Task GetChannelAvatarAsync(HttpClient httpClient, string channelId, CancellationToken cancellationToken = default) { ArgumentNullException.ThrowIfNull(httpClient); if (string.IsNullOrWhiteSpace(channelId)) @@ -165,10 +165,10 @@ private static bool IsSupportedYoutubePath(string path) => // 1) fetch youtube homepage to extract INNERTUBE_API_KEY and client version using HttpRequestMessage homeReq = new(HttpMethod.Get, "https://www.youtube.com"); homeReq.Headers.UserAgent.ParseAdd(BrowserUserAgent); - using HttpResponseMessage homeResp = await httpClient.SendAsync(homeReq).ConfigureAwait(false); + using HttpResponseMessage homeResp = await httpClient.SendAsync(homeReq, cancellationToken).ConfigureAwait(false); if (!homeResp.IsSuccessStatusCode) return null; - string homeHtml = await homeResp.Content.ReadAsStringAsync().ConfigureAwait(false); + string homeHtml = await homeResp.Content.ReadAsStringAsync(cancellationToken).ConfigureAwait(false); static string? ExtractValue(string hay, string marker) { @@ -201,10 +201,10 @@ private static bool IsSupportedYoutubePath(string path) => }; req.Headers.UserAgent.ParseAdd(BrowserUserAgent); - using HttpResponseMessage resp = await httpClient.SendAsync(req).ConfigureAwait(false); + using HttpResponseMessage resp = await httpClient.SendAsync(req, cancellationToken).ConfigureAwait(false); if (!resp.IsSuccessStatusCode) return null; - string body = await resp.Content.ReadAsStringAsync().ConfigureAwait(false); + string body = await resp.Content.ReadAsStringAsync(cancellationToken).ConfigureAwait(false); using JsonDocument doc = JsonDocument.Parse(body); JsonElement root = doc.RootElement; @@ -266,9 +266,12 @@ static bool TryTraverse(JsonElement el, string[] path, out JsonElement result) string? best = BestFromSources(thumbs2); if (!string.IsNullOrWhiteSpace(best)) return best; } - return null; } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + throw; + } catch (Exception) { return null;