feat: determine if notification is for stream or not

src/src/StevesBot.Webhook/Program.cs

@@ -1,18 +1,50 @@
+using System.Text.RegularExpressions;
+
 var builder = WebApplication.CreateBuilder(args);
 
+builder.Services.Configure<HostOptions>(
+  static options => options.ShutdownTimeout = TimeSpan.FromSeconds(30)
+);
+
 builder.Services
   .AddOptionsWithValidateOnStart<SubscriptionOptions>()
   .BindConfiguration(nameof(SubscriptionOptions))
   .ValidateDataAnnotations();
 
+builder.Services
+  .AddOptionsWithValidateOnStart<YouTubeClientOptions>()
+  .BindConfiguration(nameof(YouTubeClientOptions))
+  .ValidateDataAnnotations();
+
+builder.Services
+  .AddOptionsWithValidateOnStart<PubSubClientOptions>()
+  .BindConfiguration(nameof(PubSubClientOptions))
+  .ValidateDataAnnotations();
+
 builder.Services
   .AddHttpClient<IPubSubClient, PubSubClient>(
-    static c => c.BaseAddress = new("https://pubsubhubbub.appspot.com")
+    static (sp, c) =>
+    {
+      var options = sp.GetRequiredService<IOptions<PubSubClientOptions>>().Value;
+      c.BaseAddress = new(options.BaseUrl);
+    }
+  )
+  .AddStandardResilienceHandler();
+
+builder.Services
+  .AddHttpClient<IYouTubeDataApiClient, YouTubeDataApiClient>(
+    static (sp, c) =>
+    {
+      var options = sp.GetRequiredService<IOptions<YouTubeClientOptions>>().Value;
+      c.BaseAddress = new(options.BaseUrl);
+    }
   )
   .AddStandardResilienceHandler();
 
 builder.Services.AddOpenApi();
 
+builder.Services.AddSingleton(TimeProvider.System);
+builder.Services.AddSingleton<ConcurrentQueue<SubscribeTask>>();
 builder.Services.AddHostedService<SubscriptionWorker>();
 
 var app = builder.Build();
@@ -22,17 +54,52 @@ if (app.Environment.IsDevelopment())
   app.MapOpenApi();
 }
 
-app.UseHttpsRedirection();
+// app.UseHttpsRedirection();
 
 const string ytCallback = "yt-callback";
 
 app.MapGet(ytCallback, VerifySubscriptionHandler.HandleAsync);
 
 // TODO: Implement logic to do the following:
-// - extract video id from notification
-// - identify video as a stream or not
 // - if is stream create discord message
-// - if not then just log a message
-app.MapPost(ytCallback, static () => "hello");
+app.MapPost(ytCallback, static async (HttpContext context, [FromServices] ILogger<Program> logger, [FromServices] IYouTubeDataApiClient youTubeDataApiClient) =>
+{
+  var body = "";
+  using StreamReader stream = new(context.Request.Body);
+  body = await stream.ReadToEndAsync();
+
+  var videoIdRegex = VideoIdRegex();
+  var match = videoIdRegex.Match(body);
+
+  if (match.Success is false)
+  {
+    logger.LogWarning("No video ID found in the request body.");
+  }
+
+  var videoId = match.Groups[1].Value;
+  var video = await youTubeDataApiClient.GetVideoByIdAsync(videoId, ["liveStreamingDetails"]);
+
+  if (video is null)
+  {
+    logger.LogWarning("Video with ID {VideoId} not found.", videoId);
+    return Results.NotFound();
+  }
+
+  if (video.IsStream is false)
+  {
+    logger.LogInformation("Video ID {VideoId} is not a live stream.", videoId);
+    return Results.Ok();
+  }
+
+  logger.LogInformation("Video ID {VideoId} is a live stream.", videoId);
+
+  return Results.Ok();
+});
 
 app.Run();
+
+internal partial class Program
+{
+  [GeneratedRegex(@"<yt:videoId>(.*?)</yt:videoId>")]
+  private static partial Regex VideoIdRegex();
+}

src/src/StevesBot.Webhook/Usings.cs

@@ -1,7 +1,10 @@
+global using System.Collections.Concurrent;
 global using System.ComponentModel.DataAnnotations;
+global using System.Text.Json.Serialization;
 
 global using Microsoft.AspNetCore.Mvc;
 global using Microsoft.Extensions.Options;
 
-global using StevesBot.Webhook;
 global using StevesBot.Webhook.YouTube;
+global using StevesBot.Webhook.YouTube.Data;
+global using StevesBot.Webhook.YouTube.Tasks;

src/src/StevesBot.Webhook/YouTube/Data/IYouTubeDataApiClient.cs

@@ -0,0 +1,10 @@
+namespace StevesBot.Webhook.YouTube.Data;
+
+internal interface IYouTubeDataApiClient
+{
+  Task<YouTubeVideo?> GetVideoByIdAsync(
+    string videoId,
+    string[]? part = null,
+    CancellationToken cancellationToken = default
+  );
+}

src/src/StevesBot.Webhook/YouTube/Data/YouTubeDataApiClient.cs

@@ -0,0 +1,54 @@
+using Microsoft.AspNetCore.WebUtilities;
+
+namespace StevesBot.Webhook.YouTube.Data;
+
+internal sealed class YouTubeDataApiClient(
+  HttpClient httpClient,
+  ILogger<YouTubeDataApiClient> logger,
+  IOptions<YouTubeClientOptions> options
+) : IYouTubeDataApiClient
+{
+  private const string VideosEndpoint = "videos";
+  private readonly HttpClient _httpClient = httpClient;
+  private readonly ILogger<YouTubeDataApiClient> _logger = logger;
+  private readonly YouTubeClientOptions _options = options.Value;
+
+  public async Task<YouTubeVideo?> GetVideoByIdAsync(
+    string videoId,
+    string[]? part = null,
+    CancellationToken cancellationToken = default
+  )
+  {
+    var queryParams = new Dictionary<string, string?>
+    {
+      ["id"] = videoId,
+      ["key"] = _options.ApiKey,
+    };
+
+    if (part is not null && part.Length > 0)
+    {
+      queryParams["part"] = string.Join(',', part);
+    }
+
+    var requestEndpoint = QueryHelpers.AddQueryString(VideosEndpoint, queryParams);
+    var requestUri = new Uri(requestEndpoint, UriKind.Relative);
+    var response = await _httpClient.GetAsync(requestUri, cancellationToken);
+
+    if (response.IsSuccessStatusCode is false)
+    {
+      var content = await response.Content.ReadAsStringAsync(cancellationToken);
+
+      _logger.LogDebug(
+        "Failed to get video by ID {VideoId} from YouTube Data API. Status code: {StatusCode}, Content: {Content}",
+        videoId,
+        response.StatusCode,
+        content
+      );
+
+      return null;
+    }
+
+    var responseContent = await response.Content.ReadFromJsonAsync<YouTubeVideoListResponse>(cancellationToken);
+    return responseContent?.Items.FirstOrDefault(i => i.Id == videoId);
+  }
+}

src/src/StevesBot.Webhook/YouTube/Data/YouTubeLiveStreamingDetails.cs

@@ -0,0 +1,24 @@
+using System.Text.Json.Serialization;
+
+namespace StevesBot.Webhook.YouTube.Data;
+
+internal sealed record YouTubeLiveStreamingDetails
+{
+  [JsonPropertyName("actualStartTime")]
+  public DateTimeOffset? ActualStartTime { get; init; }
+
+  [JsonPropertyName("actualEndTime")]
+  public DateTimeOffset? ActualEndTime { get; init; }
+
+  [JsonPropertyName("scheduledStartTime")]
+  public DateTimeOffset? ScheduledStartTime { get; init; }
+
+  [JsonPropertyName("scheduledEndTime")]
+  public DateTimeOffset? ScheduledEndTime { get; init; }
+
+  [JsonPropertyName("concurrentViewers")]
+  public ulong? ConcurrentViewers { get; init; }
+
+  [JsonPropertyName("activeLiveChatId")]
+  public string? ActiveLiveChatId { get; init; }
+}

src/src/StevesBot.Webhook/YouTube/Data/YouTubePageInfo.cs

@@ -0,0 +1,12 @@
+using System.Text.Json.Serialization;
+
+namespace StevesBot.Webhook.YouTube.Data;
+
+internal sealed record YouTubePageInfo
+{
+  [JsonPropertyName("totalResults")]
+  public int TotalResults { get; init; }
+
+  [JsonPropertyName("resultsPerPage")]
+  public int ResultsPerPage { get; init; }
+}

src/src/StevesBot.Webhook/YouTube/Data/YouTubeVideo.cs

@@ -0,0 +1,12 @@
+namespace StevesBot.Webhook.YouTube.Data;
+
+internal sealed record YouTubeVideo
+{
+  [JsonPropertyName("id")]
+  public string Id { get; init; } = string.Empty;
+
+  [JsonPropertyName("liveStreamingDetails")]
+  public YouTubeLiveStreamingDetails? LiveStreamingDetails { get; init; }
+
+  public bool IsStream => LiveStreamingDetails is not null;
+}

src/src/StevesBot.Webhook/YouTube/Data/YouTubeVideoListResponse.cs

@@ -0,0 +1,12 @@
+using System.Text.Json.Serialization;
+
+namespace StevesBot.Webhook.YouTube.Data;
+
+internal sealed record YouTubeVideoListResponse
+{
+  [JsonPropertyName("items")]
+  public YouTubeVideo[] Items { get; init; } = [];
+
+  [JsonPropertyName("pageInfo")]
+  public YouTubePageInfo PageInfo { get; init; } = new YouTubePageInfo();
+}

src/src/StevesBot.Webhook/YouTube/PubSubClientOptions.cs

@@ -0,0 +1,6 @@
+namespace StevesBot.Webhook.YouTube;
+
+internal sealed class PubSubClientOptions
+{
+  public string BaseUrl { get; init; } = string.Empty;
+}

src/src/StevesBot.Webhook/YouTube/SubscriptionWorker.cs

@@ -3,15 +3,29 @@ namespace StevesBot.Webhook.YouTube;
 internal sealed class SubscriptionWorker(
   ILogger<SubscriptionWorker> logger,
   IOptions<SubscriptionOptions> options,
-  IPubSubClient pubSubClient
-) : IHostedLifecycleService
+  IPubSubClient pubSubClient,
+  ConcurrentQueue<SubscribeTask> subscriptionQueue,
+  TimeProvider timeProvider
+) : IHostedLifecycleService, IDisposable
 {
   private readonly ILogger<SubscriptionWorker> _logger = logger;
   private readonly SubscriptionOptions _options = options.Value;
   private readonly IPubSubClient _pubSubClient = pubSubClient;
+  private readonly ConcurrentQueue<SubscribeTask> _subscriptionQueue = subscriptionQueue;
+  private readonly TimeProvider _timeProvider = timeProvider;
+  private ITimer? _subscriptionTimer;
 
   public Task StartAsync(CancellationToken cancellationToken)
   {
+    _logger.LogInformation("Starting subscription worker");
+
+    _subscriptionTimer = _timeProvider.CreateTimer(
+      callback: async _ => await ProcessSubscriptionQueueAsync(cancellationToken),
+      state: null,
+      dueTime: TimeSpan.FromSeconds(0),
+      period: TimeSpan.FromDays(1)
+    );
+
     return Task.CompletedTask;
   }
 
@@ -42,6 +56,7 @@ internal sealed class SubscriptionWorker(
 
   public Task StopAsync(CancellationToken cancellationToken)
   {
+    _subscriptionTimer?.Change(Timeout.InfiniteTimeSpan, TimeSpan.Zero);
     return Task.CompletedTask;
   }
 
@@ -54,4 +69,52 @@ internal sealed class SubscriptionWorker(
   {
     return Task.CompletedTask;
   }
+
+  private async Task ProcessSubscriptionQueueAsync(CancellationToken cancellationToken)
+  {
+    while (_subscriptionQueue.TryDequeue(out var task))
+    {
+      // We will attempt to process the task
+      // once every day so we will aim to 
+      // process it two days before it expires.
+      var now = _timeProvider.GetUtcNow();
+      var dueAt = task.ExpiresAt - TimeSpan.FromDays(2);
+      var isDue = now >= dueAt;
+
+      if (isDue is false)
+      {
+        _logger.LogInformation(
+          "Skipping subscription task for topic {TopicUrl} as it is not due yet. Due at: {DueAt}, Current time: {CurrentTime}",
+          task.TopicUrl,
+          dueAt,
+          now
+        );
+
+        _subscriptionQueue.Enqueue(task);
+
+        continue;
+      }
+
+      _logger.LogInformation("Processing subscription task for topic: {TopicUrl}", task.TopicUrl);
+
+      var isSubscribed = await _pubSubClient.SubscribeAsync(
+        task.CallbackUrl,
+        task.TopicUrl,
+        cancellationToken
+      );
+
+      if (isSubscribed)
+      {
+        _logger.LogInformation("Successfully subscribed to topic: {TopicUrl}", task.TopicUrl);
+        return;
+      }
+
+      _logger.LogWarning("Failed to subscribe to topic: {TopicUrl}", task.TopicUrl);
+    }
+  }
+
+  public void Dispose()
+  {
+    _subscriptionTimer?.Dispose();
+  }
 }

src/src/StevesBot.Webhook/YouTube/Tasks/SubscribeTask.cs

@@ -1,6 +1,8 @@
 namespace StevesBot.Webhook.YouTube.Tasks;
 
-internal sealed record SubscribeTask : SubscriptionTask
+internal sealed record SubscribeTask
 {
+  public string CallbackUrl { get; init; } = string.Empty;
+  public string TopicUrl { get; init; } = string.Empty;
   public DateTimeOffset ExpiresAt { get; init; }
 }

src/src/StevesBot.Webhook/YouTube/Tasks/SubscriptionTask.cs

@@ -1,7 +0,0 @@
-namespace StevesBot.Webhook.YouTube.Tasks;
-
-internal abstract record SubscriptionTask
-{
-  public string CallbackUrl { get; init; } = string.Empty;
-  public string TopicUrl { get; init; } = string.Empty;
-}

src/src/StevesBot.Webhook/YouTube/Tasks/UnsubscribeTask.cs

@@ -1,5 +0,0 @@
-namespace StevesBot.Webhook.YouTube.Tasks;
-
-internal sealed record UnsubscribeTask : SubscriptionTask
-{
-}

src/src/StevesBot.Webhook/YouTube/VerifySubscriptionHandler.cs

@@ -9,7 +9,8 @@ internal static class VerifySubscriptionHandler
     [FromQuery(Name = "hub.challenge")] string? challenge,
     [FromQuery(Name = "hub.lease_seconds")] string? leaseSeconds,
     [FromServices] IOptions<SubscriptionOptions> subOptions,
-    [FromServices] ILogger<Program> logger
+    [FromServices] ILogger<Program> logger,
+    [FromServices] ConcurrentQueue<SubscribeTask> subscriptionQueue
   )
   {
     if (mode is "denied")
@@ -26,23 +27,31 @@ internal static class VerifySubscriptionHandler
 
     if (mode is "subscribe")
     {
-      // TODO: If it is a subscription request
-      // we need to queue up a resubscription
-      // request to be executed just before
-      // the hub.lease expires.
       logger.LogInformation(
         "Received subscription request for topic: {Topic}, challenge: {Challenge}, lease: {LeaseSeconds}",
         topic,
         challenge,
         leaseSeconds
       );
+
+      if (string.IsNullOrWhiteSpace(leaseSeconds) || !long.TryParse(leaseSeconds, out var parsedSeconds))
+      {
+        logger.LogWarning("Invalid or missing lease_seconds parameter: {LeaseSeconds}", leaseSeconds);
+        return Results.BadRequest("Invalid lease_seconds parameter");
+      }
+
+      var task = new SubscribeTask
+      {
+        CallbackUrl = subOptions.Value.CallbackUrl,
+        TopicUrl = topic,
+        ExpiresAt = DateTime.UtcNow.AddSeconds(parsedSeconds),
+      };
+
+      subscriptionQueue.Enqueue(task);
     }
 
     if (mode is "unsubscribe")
     {
-      // TODO: If it is a unsubscription request
-      // we need to queue up an ubsubscription
-      // request to be executed immediately
       logger.LogInformation(
         "Received unsubscription request for topic: {Topic}, challenge: {Challenge}",
         topic,

src/src/StevesBot.Webhook/YouTube/YouTubeClientOptions.cs

@@ -0,0 +1,10 @@
+namespace StevesBot.Webhook.YouTube;
+
+internal sealed record YouTubeClientOptions
+{
+  [Required]
+  public string BaseUrl { get; init; } = string.Empty;
+
+  [Required]
+  public string ApiKey { get; init; } = string.Empty;
+}

src/src/StevesBot.Webhook/appsettings.Example.json

@@ -1,6 +1,20 @@
 {
+  "Logging": {
+    "LogLevel": {
+      "Default": "Information",
+      "Microsoft.AspNetCore": "Warning"
+    }
+  },
+  "AllowedHosts": "*",
+  "YouTubeClientOptions": {
+    "BaseUrl": "BaseUrl",
+    "ApiKey": "ApiKey"
+  },
   "SubscriptionOptions": {
     "CallbackUrl": "CallbackUrl",
     "TopicUrl": "TopicUrl"
+  },
+  "PubSubClientOptions": {
+    "BaseUrl": "BaseUrl"
   }
 }