From 1dbda14335421453c89b8697a4579e1f1169bfbe Mon Sep 17 00:00:00 2001
From: Kavin <20838718+FireMasterK@users.noreply.github.com>
Date: Thu, 6 Jul 2023 14:54:45 +0100
Subject: [PATCH 1/2] Use an upsert statement for handling pubsub.

---
 .../java/me/kavin/piped/utils/PubSubHelper.java | 17 +++++++----------
 1 file changed, 7 insertions(+), 10 deletions(-)

diff --git a/src/main/java/me/kavin/piped/utils/PubSubHelper.java b/src/main/java/me/kavin/piped/utils/PubSubHelper.java
index 23976b74..e1b66066 100644
--- a/src/main/java/me/kavin/piped/utils/PubSubHelper.java
+++ b/src/main/java/me/kavin/piped/utils/PubSubHelper.java
@@ -57,17 +57,14 @@ public static void subscribePubSub(String channelId) throws IOException {
     }
 
     public static void updatePubSub(String channelId) {
-        var pubsub = DatabaseHelper.getPubSubFromId(channelId);
         try (StatelessSession s = DatabaseSessionFactory.createStatelessSession()) {
-            s.beginTransaction();
-            if (pubsub == null) {
-                pubsub = new PubSub(channelId, System.currentTimeMillis());
-                s.insert(pubsub);
-            } else {
-                pubsub.setSubbedAt(System.currentTimeMillis());
-                s.update(pubsub);
-            }
-            s.getTransaction().commit();
+            var tr = s.beginTransaction();
+            s.createNativeMutationQuery("INSERT INTO pubsub (id, subbed_at) VALUES (?, ?) " +
+                            "ON CONFLICT (id) DO UPDATE SET subbed_at = excluded.subbed_at")
+                    .setParameter(1, channelId)
+                    .setParameter(2, System.currentTimeMillis())
+                    .executeUpdate();
+            tr.commit();
         }
     }
 }

From f9289a05d12f60f761ba53316963ed416435387d Mon Sep 17 00:00:00 2001
From: Kavin <20838718+FireMasterK@users.noreply.github.com>
Date: Thu, 6 Jul 2023 14:56:28 +0100
Subject: [PATCH 2/2] Use a limited threadpool for handling pubsub requests.

---
 .../java/me/kavin/piped/server/ServerLauncher.java    | 11 +++++++----
 1 file changed, 7 insertions(+), 4 deletions(-)

diff --git a/src/main/java/me/kavin/piped/server/ServerLauncher.java b/src/main/java/me/kavin/piped/server/ServerLauncher.java
index 44e83ada..4626c089 100644
--- a/src/main/java/me/kavin/piped/server/ServerLauncher.java
+++ b/src/main/java/me/kavin/piped/server/ServerLauncher.java
@@ -76,18 +76,21 @@ AsyncServlet mainServlet(Executor executor) {
                 .map(GET, "/webhooks/pubsub", AsyncServlet.ofBlocking(executor, request -> {
                     var topic = request.getQueryParameter("hub.topic");
                     if (topic != null)
-                        Multithreading.runAsync(() -> {
+                        Multithreading.runAsyncLimited(() -> {
                             String channelId = StringUtils.substringAfter(topic, "channel_id=");
                             PubSubHelper.updatePubSub(channelId);
                         });
-                    return HttpResponse.ok200().withPlainText(Objects.requireNonNull(request.getQueryParameter("hub.challenge")));
+
+                    var challenge = request.getQueryParameter("hub.challenge");
+                    return HttpResponse.ok200()
+                            .withPlainText(Objects.requireNonNullElse(challenge, "ok"));
                 })).map(POST, "/webhooks/pubsub", AsyncServlet.ofBlocking(executor, request -> {
                     try {
 
                         SyndFeed feed = new SyndFeedInput().build(
                                 new InputSource(new ByteArrayInputStream(request.loadBody().getResult().asArray())));
 
-                        Multithreading.runAsync(() -> {
+                        Multithreading.runAsyncLimited(() -> {
                             for (var entry : feed.getEntries()) {
                                 String url = entry.getLinks().get(0).getHref();
                                 String videoId = StringUtils.substring(url, -11);
@@ -95,7 +98,7 @@ AsyncServlet mainServlet(Executor executor) {
                                     if (DatabaseHelper.doesVideoExist(s, videoId))
                                         continue;
                                 }
-                                Multithreading.runAsync(() -> {
+                                Multithreading.runAsyncLimited(() -> {
                                     try {
                                         Sentry.setExtra("videoId", videoId);
                                         var extractor = YOUTUBE_SERVICE.getStreamExtractor("https://youtube.com/watch?v=" + videoId);