kavin/Piped-Backend
1
0
Fork 0
mirror of https://github.com/TeamPiped/Piped-Backend.git synced 2024-12-14 06:10:29 +05:30
Code Issues Projects Releases Wiki Activity

Merge pull request #372 from TeamPiped/pubsub-opt

Browse Source 
Optimize pubsub subscription query
This commit is contained in:
Kavin 2022-09-17 20:33:11 +05:30 committed by GitHub
commit bfa7bfa79a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 24 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

60
src/main/java/me/kavin/piped/Main.java
View File

@ -2,10 +2,10 @@ package me.kavin.piped;
import io.activej.inject.Injector; import io.activej.inject.Injector;
import jakarta.persistence.criteria.CriteriaBuilder; import jakarta.persistence.criteria.CriteriaBuilder;
import jakarta.persistence.criteria.CriteriaQuery;
import me.kavin.piped.consts.Constants; import me.kavin.piped.consts.Constants;
import me.kavin.piped.utils.*; import me.kavin.piped.utils.*;
import me.kavin.piped.utils.obj.db.*; import me.kavin.piped.utils.obj.db.PlaylistVideo;
import me.kavin.piped.utils.obj.db.Video;
import org.hibernate.Session; import org.hibernate.Session;
import org.hibernate.StatelessSession; import org.hibernate.StatelessSession;
import org.schabi.newpipe.extractor.NewPipe; import org.schabi.newpipe.extractor.NewPipe;
@ -17,6 +17,7 @@ import org.schabi.newpipe.extractor.services.youtube.extractors.YoutubeStreamExt
import java.io.IOException; import java.io.IOException;
import java.util.*; import java.util.*;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
public class Main { public class Main {
@ -64,41 +65,28 @@ public class Main {
public void run() { public void run() {
try (StatelessSession s = DatabaseSessionFactory.createStatelessSession()) { try (StatelessSession s = DatabaseSessionFactory.createStatelessSession()) {
CriteriaBuilder cb = s.getCriteriaBuilder(); final Random random = new Random();
CriteriaQuery<PubSub> criteria = cb.createQuery(PubSub.class);
var root = criteria.from(PubSub.class);
var userSq = criteria.subquery(User.class);
var userRoot = userSq.from(User.class);
userSq.select(userRoot.get("subscribed_ids"));
var subquery = criteria.subquery(UnauthenticatedSubscription.class);
var subRoot = subquery.from(UnauthenticatedSubscription.class);
subquery.select(subRoot.get("id"))
.where(cb.gt(subRoot.get("subscribedAt"), System.currentTimeMillis() - TimeUnit.DAYS.toMillis(Constants.SUBSCRIPTIONS_EXPIRY)));
criteria.select(root)
.where(cb.and(
cb.lessThan(root.get("subbedAt"), System.currentTimeMillis() - TimeUnit.DAYS.toMillis(4)),
cb.or(
root.get("id").in(userSq),
root.get("id").in(subquery)
)
)
);
List<PubSub> pubSubList = s.createQuery(criteria).list(); s.createNativeQuery("SELECT id FROM pubsub WHERE subbed_at < :subbedTime AND id IN (" +
"SELECT DISTINCT channel FROM users_subscribed" +
Collections.shuffle(pubSubList); " UNION " +
"SELECT id FROM unauthenticated_subscriptions WHERE subscribed_at > :unauthSubbed" +
pubSubList.stream().parallel() ")", String.class)
.forEach(pubSub -> { .setParameter("subbedTime", System.currentTimeMillis() - TimeUnit.DAYS.toMillis(4))
if (pubSub != null) .setParameter("unauthSubbed", System.currentTimeMillis() - TimeUnit.DAYS.toMillis(Constants.SUBSCRIPTIONS_EXPIRY))
Multithreading.runAsyncLimitedPubSub(() -> { .stream()
try { .filter(Objects::nonNull)
ResponseHelper.subscribePubSub(pubSub.getId()); .collect(Collectors.toUnmodifiableSet())
} catch (IOException e) { .stream()
ExceptionHandler.handle(e); .sorted(Comparator.comparingInt(o -> random.nextInt()))
} .parallel()
}); .forEach(id -> Multithreading.runAsyncLimitedPubSub(() -> {
}); try {
ResponseHelper.subscribePubSub(id);
} catch (IOException e) {
ExceptionHandler.handle(e);
}
}));
} catch (Exception e) { } catch (Exception e) {
e.printStackTrace(); e.printStackTrace();