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 23976b7..e1b6606 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 class PubSubHelper { } 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 44e83ad..4626c08 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 @@ public class ServerLauncher extends MultithreadedHttpServerLauncher { .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 @@ public class ServerLauncher extends MultithreadedHttpServerLauncher { 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);