diff --git a/modules/BlueskyGephi/pom.xml b/modules/BlueskyGephi/pom.xml index 3dc644532..069f6e039 100644 --- a/modules/BlueskyGephi/pom.xml +++ b/modules/BlueskyGephi/pom.xml @@ -9,7 +9,7 @@ fr.totetmatt bluesky-gephi - 0.2.1 + 0.2.2 nbm Bluesky Gephi diff --git a/modules/BlueskyGephi/src/main/java/fr/totetmatt/blueskygephi/BlueskyGephi.java b/modules/BlueskyGephi/src/main/java/fr/totetmatt/blueskygephi/BlueskyGephi.java index dcffd5525..1ad119662 100644 --- a/modules/BlueskyGephi/src/main/java/fr/totetmatt/blueskygephi/BlueskyGephi.java +++ b/modules/BlueskyGephi/src/main/java/fr/totetmatt/blueskygephi/BlueskyGephi.java @@ -1,20 +1,22 @@ package fr.totetmatt.blueskygephi; import fr.totetmatt.blueskygephi.atproto.AtClient; -import fr.totetmatt.blueskygephi.atproto.response.AppBskyGraphGetFollowers; -import fr.totetmatt.blueskygephi.atproto.response.AppBskyGraphGetFollows; -import fr.totetmatt.blueskygephi.atproto.response.AppBskyGraphGetList; import fr.totetmatt.blueskygephi.atproto.response.common.Identity; import java.awt.Color; +import java.util.ArrayList; import java.util.HashSet; import java.util.List; import java.util.Optional; import java.util.Set; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.function.LongConsumer; import java.util.logging.Logger; import java.util.prefs.Preferences; import java.util.stream.Collectors; import java.util.stream.Stream; import org.gephi.graph.api.Edge; +import org.gephi.graph.api.Graph; import org.gephi.graph.api.GraphController; import org.gephi.graph.api.GraphModel; import org.gephi.graph.api.Node; @@ -51,6 +53,15 @@ public class BlueskyGephi { private AtClient client; private GraphModel graphModel; + // Bounded pool so pasting many handles can't spawn an unbounded number of + // threads all contending on the graph write-lock and hammering the API. + private final ExecutorService executor = Executors.newFixedThreadPool(2, r -> { + Thread t = new Thread(r); + t.setDaemon(true); + t.setName("[Bsky] worker"); + return t; + }); + public BlueskyGephi() { } @@ -161,87 +172,165 @@ private Edge createEdge(Node source, Node target) { } private void fetchFollowerFollowsFromActor(String actor, List listInit, boolean isFollowsActive, boolean isFollowersActive, boolean isDeepSearch) { - // To avoid locking Gephi UI - Thread t = new Thread() { - private ProgressTicket progressTicket; - Set foaf = new HashSet<>(); + // Run on a bounded background pool to keep the Gephi UI responsive. + executor.submit(new FetchTask(actor, listInit, isFollowsActive, isFollowersActive, isDeepSearch)); + } - private void process(String actor, boolean isDeepSearch, Optional limitCrawl) { + /** + * A single fetch job. Reads the network for one actor (and, when deep + * search or a list is involved, for the discovered friends-of-a-friend) + * and writes it into the graph. + */ + private final class FetchTask implements Runnable { + + private final String actor; + private final List listInit; + private final boolean isFollowsActive; + private final boolean isFollowersActive; + private final boolean isDeepSearch; + + private final Set foaf = new HashSet<>(); + private volatile boolean cancelled = false; + private volatile Thread worker; + private ProgressTicket progressTicket; + private LongConsumer onWaiting; + + FetchTask(String actor, List listInit, boolean isFollowsActive, boolean isFollowersActive, boolean isDeepSearch) { + this.actor = actor; + this.listInit = listInit; + this.isFollowsActive = isFollowsActive; + this.isFollowersActive = isFollowersActive; + this.isDeepSearch = isDeepSearch; + } - try { - if (isFollowsActive) { - List responses = client.appBskyGraphGetFollows(actor, limitCrawl); + private boolean shouldStop() { + return cancelled || Thread.currentThread().isInterrupted(); + } - for (var response : responses) { - graphModel.getGraph().writeLock(); - Identity subject = response.getSubject(); + private void process(String actor, boolean isDeepSearch, Optional limitCrawl) { + if (client == null || graphModel == null) { + return; + } + try { + if (isFollowsActive && !shouldStop()) { + // Each page is written as it arrives; pagination stops when the + // worker is interrupted (cancellation). + client.appBskyGraphGetFollows(actor, limitCrawl, response -> { + if (shouldStop()) { + return; + } + Identity subject = response.getSubject(); + if (subject == null || subject.getDid() == null) { + return; + } + // Capture the graph once so lock/unlock always target the + // same instance, and release it in finally so an exception + // (bad data, API error, cancellation) can never leak the + // write-lock and freeze Gephi. + Graph graph = graphModel.getGraph(); + graph.writeLock(); + try { Node source = createNode(subject); source.setColor(Color.GREEN); - for (var follow : response.getFollows()) { - if (isDeepSearch) { - foaf.add(follow.getDid()); + List follows = response.getFollows(); + if (follows != null) { + for (var follow : follows) { + if (follow == null || follow.getDid() == null) { + continue; + } + if (isDeepSearch) { + foaf.add(follow.getDid()); + } + Node target = createNode(follow); + createEdge(source, target); } - Node target = createNode(follow); - createEdge(source, target); } - graphModel.getGraph().writeUnlock(); - + } finally { + graph.writeUnlock(); } + }, onWaiting); + } - } - - if (isFollowersActive) { - List responses = client.appBskyGraphGetFollowers(actor, limitCrawl); - - for (var response : responses) { - graphModel.getGraph().writeLock(); - Identity subject = response.getSubject(); + if (isFollowersActive && !shouldStop()) { + client.appBskyGraphGetFollowers(actor, limitCrawl, response -> { + if (shouldStop()) { + return; + } + Identity subject = response.getSubject(); + if (subject == null || subject.getDid() == null) { + return; + } + Graph graph = graphModel.getGraph(); + graph.writeLock(); + try { Node target = createNode(subject); target.setColor(Color.GREEN); - for (var follower : response.getFollowers()) { - if (isDeepSearch) { - foaf.add(follower.getDid()); + List followers = response.getFollowers(); + if (followers != null) { + for (var follower : followers) { + if (follower == null || follower.getDid() == null) { + continue; + } + if (isDeepSearch) { + foaf.add(follower.getDid()); + } + Node source = createNode(follower); + createEdge(source, target); } - Node source = createNode(follower); - createEdge(source, target); } - graphModel.getGraph().writeUnlock(); + } finally { + graph.writeUnlock(); } - - } - } catch (Exception e) { + }, onWaiting); + } + } catch (Exception e) { + if (!cancelled) { Exceptions.printStackTrace(e); - } finally { } } + } - @Override - public void run() { - - if (actor != null) { - this.setName("[Bsky] fetching" + actor); + @Override + public void run() { + worker = Thread.currentThread(); + final String taskName = (actor != null) ? "[Bsky] fetching " + actor : "[Bsky] fetching List"; + worker.setName(taskName); + + progressTicket = Lookup.getDefault() + .lookup(ProgressTicketProvider.class) + .createTicket(taskName, () -> { + cancelled = true; + Thread w = worker; + if (w != null) { + w.interrupt(); + } + return true; + }); + // Surface the client's rate-limit backoff as a live countdown in the progress bar. + onWaiting = remainingMillis -> { + if (remainingMillis <= 0L) { + Progress.setDisplayName(progressTicket, taskName); } else { - this.setName("[Bsky] fetching List"); + long seconds = (remainingMillis + 999L) / 1000L; + Progress.setDisplayName(progressTicket, taskName + " - rate limited, retrying in " + seconds + "s"); } - progressTicket = Lookup.getDefault() - .lookup(ProgressTicketProvider.class) - .createTicket(this.getName(), () -> { - interrupt(); - Progress.finish(progressTicket); - return true; - }); + }; + try { Progress.start(progressTicket); Progress.switchToIndeterminate(progressTicket); if (listInit != null) { - this.foaf.addAll(listInit); + foaf.addAll(listInit); } if (actor != null) { process(actor, isDeepSearch, Optional.empty()); } - if (listInit != null || isDeepSearch) { + if ((listInit != null || isDeepSearch) && !shouldStop()) { Progress.switchToDeterminate(progressTicket, foaf.size()); for (var foafActor : foaf) { + if (shouldStop()) { + break; + } Progress.setDisplayName(progressTicket, "[Bsky] fetching " + actor + " n+1 > " + foafActor); if (getIsLimitCrawlActive()) { process(foafActor, false, Optional.of(getLimitCrawl())); @@ -249,20 +338,26 @@ public void run() { process(foafActor, false, Optional.empty()); } Progress.progress(progressTicket); - } } + } finally { Progress.finish(progressTicket); } - }; - t.start(); - + } } private Stream manageList(String listId) { - List list = client.appBskyGraphGetList(listId); - return list.stream().flatMap(x -> x.getItems().stream().map(y -> y.getSubject().getDid())); - + List dids = new ArrayList<>(); + client.appBskyGraphGetList(listId, page -> { + if (page.getItems() != null) { + for (var item : page.getItems()) { + if (item != null && item.getSubject() != null && item.getSubject().getDid() != null) { + dids.add(item.getSubject().getDid()); + } + } + } + }, null); + return dids.stream(); } private void initGraphTable() { @@ -273,28 +368,50 @@ private void initGraphTable() { } public void fetchFollowerFollowsFromActors(List actors, boolean isFollowsActive, boolean isFollowersActive, boolean isBlocksActive) { + if (client == null) { + logger.warning("Not connected to Bluesky. Please connect before fetching."); + return; + } ensureProject(); graphModel = Lookup.getDefault().lookup(GraphController.class).getGraphModel(); initGraphTable(); - actors.stream().forEach(actor -> fetchFollowerFollowsFromActor(actor, null, isFollowsActive, isFollowersActive, getIsDeepSearch())); + actors.stream() + .map(String::trim) + .filter(actor -> !actor.isEmpty()) + .forEach(actor -> fetchFollowerFollowsFromActor(actor, null, isFollowsActive, isFollowersActive, getIsDeepSearch())); } public void fetchFollowerFollowsFromActors(List actors) { + if (client == null) { + logger.warning("Not connected to Bluesky. Please connect before fetching."); + return; + } ensureProject(); graphModel = Lookup.getDefault().lookup(GraphController.class).getGraphModel(); initGraphTable(); - actors - .stream() + + List cleanActors = actors.stream() + .map(String::trim) + .filter(x -> !x.isEmpty()) + .collect(Collectors.toList()); + + cleanActors.stream() .filter(x -> !x.contains("app.bsky.graph.list")) .forEach(actor -> fetchFollowerFollowsFromActor(actor, null, getIsFollowsActive(), getIsFollowersActive(), getIsDeepSearch())); - List listActor = actors - .stream() + List listIds = cleanActors.stream() .filter(x -> x.contains("app.bsky.graph.list")) - .flatMap(this::manageList) .collect(Collectors.toList()); - fetchFollowerFollowsFromActor(null, listActor, getIsFollowsActive(), getIsFollowersActive(), getIsDeepSearch()); - + if (!listIds.isEmpty()) { + // Resolving a list is itself several paged network calls, so do it + // off the EDT instead of blocking the UI thread. + executor.submit(() -> { + List listActor = listIds.stream() + .flatMap(this::manageList) + .collect(Collectors.toList()); + fetchFollowerFollowsFromActor(null, listActor, getIsFollowsActive(), getIsFollowersActive(), getIsDeepSearch()); + }); + } } } diff --git a/modules/BlueskyGephi/src/main/java/fr/totetmatt/blueskygephi/BlueskyGephiMainPanel.java b/modules/BlueskyGephi/src/main/java/fr/totetmatt/blueskygephi/BlueskyGephiMainPanel.java index 1d114dc24..5e4f9bbca 100644 --- a/modules/BlueskyGephi/src/main/java/fr/totetmatt/blueskygephi/BlueskyGephiMainPanel.java +++ b/modules/BlueskyGephi/src/main/java/fr/totetmatt/blueskygephi/BlueskyGephiMainPanel.java @@ -269,11 +269,32 @@ private void isFollowersActivatedActionPerformed(java.awt.event.ActionEvent evt) }//GEN-LAST:event_isFollowersActivatedActionPerformed private void credentialsConnectButtonActionPerformed(java.awt.event.ActionEvent evt) {//GEN-FIRST:event_credentialsConnectButtonActionPerformed - if(blueskyGephi.connect(credentialsHostField.getText(), credentialsHandleField.getText(), String.valueOf(credentialsPasswordField.getPassword()))){ - credentialsConnectButton.setBackground(Color.GREEN); - } else { - credentialsConnectButton.setBackground(Color.RED); - } + final String host = credentialsHostField.getText(); + final String handle = credentialsHandleField.getText(); + final String password = String.valueOf(credentialsPasswordField.getPassword()); + // Do the network call off the EDT so a slow or failing login never freezes the UI. + credentialsConnectButton.setEnabled(false); + credentialsConnectButton.setBackground(Color.ORANGE); + new javax.swing.SwingWorker() { + @Override + protected Boolean doInBackground() { + return blueskyGephi.connect(host, handle, password); + } + + @Override + protected void done() { + boolean connected = false; + try { + connected = get(); + } catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + } catch (java.util.concurrent.ExecutionException ex) { + Exceptions.printStackTrace(ex); + } + credentialsConnectButton.setBackground(connected ? Color.GREEN : Color.RED); + credentialsConnectButton.setEnabled(true); + } + }.execute(); }//GEN-LAST:event_credentialsConnectButtonActionPerformed private void runFetchButtonActionPerformed(java.awt.event.ActionEvent evt) {//GEN-FIRST:event_runFetchButtonActionPerformed diff --git a/modules/BlueskyGephi/src/main/java/fr/totetmatt/blueskygephi/atproto/AtClient.java b/modules/BlueskyGephi/src/main/java/fr/totetmatt/blueskygephi/atproto/AtClient.java index 4281f6743..9faed2fca 100644 --- a/modules/BlueskyGephi/src/main/java/fr/totetmatt/blueskygephi/atproto/AtClient.java +++ b/modules/BlueskyGephi/src/main/java/fr/totetmatt/blueskygephi/atproto/AtClient.java @@ -6,157 +6,273 @@ */ import com.fasterxml.jackson.databind.DeserializationFeature; import com.fasterxml.jackson.databind.ObjectMapper; -import fr.totetmatt.blueskygephi.atproto.response.AppBskyActorGetProfile; import fr.totetmatt.blueskygephi.atproto.response.AppBskyGraphGetFollowers; import fr.totetmatt.blueskygephi.atproto.response.AppBskyGraphGetFollows; import fr.totetmatt.blueskygephi.atproto.response.AppBskyGraphGetList; import fr.totetmatt.blueskygephi.atproto.response.ComAtprotoServerCreateSession; +import fr.totetmatt.blueskygephi.atproto.response.Paged; import java.io.IOException; import java.net.http.HttpClient; import java.net.http.HttpRequest; import java.net.http.HttpResponse; -import java.util.ArrayList; +import java.time.Duration; import java.util.HashMap; -import java.util.List; +import java.util.Map; import java.util.Optional; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import org.openide.util.Exceptions; +import java.util.concurrent.ThreadLocalRandom; +import java.util.function.Consumer; +import java.util.function.LongConsumer; +import java.util.logging.Logger; public class AtClient { + private static final Logger logger = Logger.getLogger(AtClient.class.getName()); + + private static final int MAX_RETRIES = 5; + private static final Duration CONNECT_TIMEOUT = Duration.ofSeconds(15); + private static final Duration REQUEST_TIMEOUT = Duration.ofSeconds(30); + // Cap on how long a single backoff can sleep. Rate-limit windows are up to + // 5 minutes, so honor a reported reset that far out but no further. + private static final long MAX_BACKOFF_MS = 300_000L; + private final ObjectMapper objectMapper = new ObjectMapper().configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false); private final AtContext context; + private final HttpClient client; + + // Mutated by refreshSession() from any worker thread, read everywhere. + private volatile ComAtprotoServerCreateSession session = null; - private final HttpClient client = HttpClient.newHttpClient(); - private ComAtprotoServerCreateSession session = null; -ExecutorService executorService = Executors.newFixedThreadPool(2); public AtClient(String host) { - context = new AtContext(host); + this(host, HttpClient.newBuilder().connectTimeout(CONNECT_TIMEOUT).build()); + } + + // Package-private: lets tests inject a mocked HttpClient without hitting the network. + AtClient(String host, HttpClient client) { + this.context = new AtContext(host); + this.client = client; } public boolean comAtprotoServerCreateSession(String identifier, String password) { try { + // Build the JSON body with the mapper so a password containing + // quotes/backslashes can't break or inject into the request. + String body = objectMapper.writeValueAsString(Map.of( + "identifier", identifier, + "password", password)); HttpRequest request = HttpRequest.newBuilder(context.getURIForLexicon("com.atproto.server.createSession")) - .POST(HttpRequest.BodyPublishers.ofString("{\"identifier\":\"" + identifier + "\",\"password\":\"" + password + "\"}")) + .POST(HttpRequest.BodyPublishers.ofString(body)) + .timeout(REQUEST_TIMEOUT) .header("Content-Type", "application/json") .build(); - var response = client.send(request, HttpResponse.BodyHandlers.ofString()); + HttpResponse response = client.send(request, HttpResponse.BodyHandlers.ofString()); + if (response.statusCode() != 200) { + logger.warning("Bluesky createSession failed (HTTP " + response.statusCode() + "): " + response.body()); + this.session = null; + return false; + } this.session = objectMapper.readValue(response.body(), ComAtprotoServerCreateSession.class); - return true; - } catch (IOException | InterruptedException e) { - throw new RuntimeException(e); + return this.session != null && this.session.getAccessJwt() != null; + } catch (IOException e) { + logger.warning("Bluesky createSession error: " + e.getMessage()); + this.session = null; + return false; + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + this.session = null; + return false; } } - private HttpRequest getRequest(String xrpcMethod, HashMap params) { - return HttpRequest - .newBuilder(context.getURIForLexicon(xrpcMethod, params)) - .GET() - .header("Authorization", "Bearer " + session.getAccessJwt()) - .build(); - } - - // Yeah, it should be generalized, and async, but it works right now so it's ok. - public List appBskyGraphGetFollowers(String actor, Optional limitCrawl) { - List pagedResponse = new ArrayList<>(); + /** + * Exchanges the (short-lived) access token for a fresh one using the refresh + * token. Synchronized so two worker threads that both hit a 401 at the same + * time don't each burn a refresh. + * + * @param staleAccessJwt the access token that just got a 401; if the current + * session no longer uses it, another thread already refreshed and we're done. + */ + private synchronized boolean refreshSession(String staleAccessJwt) { + ComAtprotoServerCreateSession current = session; + if (current == null) { + return false; + } + if (staleAccessJwt != null && !staleAccessJwt.equals(current.getAccessJwt())) { + return true; + } + if (current.getRefreshJwt() == null) { + return false; + } try { - var params = new HashMap(); - params.put("actor", actor); - params.put("limit", "100"); - int currentCrawlLoop=0; - while (limitCrawl.isEmpty() ||currentCrawlLoop < limitCrawl.get()) { - var request = getRequest("app.bsky.graph.getFollowers", params); - var response = client.send(request, HttpResponse.BodyHandlers.ofString()); - - AppBskyGraphGetFollowers objectResponse = objectMapper.readValue(response.body(), AppBskyGraphGetFollowers.class); - pagedResponse.add(objectResponse); - if (objectResponse.getCursor() == null) { - break; - } - params.put("cursor", objectResponse.getCursor()); - currentCrawlLoop++; + HttpRequest request = HttpRequest.newBuilder(context.getURIForLexicon("com.atproto.server.refreshSession")) + .POST(HttpRequest.BodyPublishers.noBody()) + .timeout(REQUEST_TIMEOUT) + .header("Authorization", "Bearer " + current.getRefreshJwt()) + .build(); + HttpResponse response = client.send(request, HttpResponse.BodyHandlers.ofString()); + if (response.statusCode() != 200) { + logger.warning("Bluesky refreshSession failed (HTTP " + response.statusCode() + ")"); + return false; } - return pagedResponse; - } catch (IOException | InterruptedException e) { - throw new RuntimeException(e); + ComAtprotoServerCreateSession refreshed = objectMapper.readValue(response.body(), ComAtprotoServerCreateSession.class); + if (refreshed == null || refreshed.getAccessJwt() == null) { + return false; + } + this.session = refreshed; + return true; + } catch (IOException e) { + logger.warning("Bluesky refreshSession error: " + e.getMessage()); + return false; + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return false; } } - public List appBskyGraphGetFollows(String actor, Optional limitCrawl) { - List pagedResponse = new ArrayList<>(); - try { - var params = new HashMap(); - params.put("actor", actor); - params.put("limit", "100"); - int currentCrawlLoop=0; - while (limitCrawl.isEmpty() || currentCrawlLoop < limitCrawl.get()) { - var request = getRequest("app.bsky.graph.getFollows", params); - var response = client.send(request, HttpResponse.BodyHandlers.ofString()); - var objectResponse = objectMapper.readValue(response.body(), AppBskyGraphGetFollows.class); - pagedResponse.add(objectResponse); - if (objectResponse.getCursor() == null) { - break; + /** + * Single entry point for authenticated GET calls: attaches the bearer token, + * refreshes once on 401 and retries, backs off and retries on 429/5xx and + * transient network errors, and deserializes a 200 body into {@code type}. + */ + private T get(String lexicon, Map params, Class type, LongConsumer onWaiting) { + boolean refreshed = false; + int attempt = 0; + while (true) { + ComAtprotoServerCreateSession current = session; + if (current == null || current.getAccessJwt() == null) { + throw new AtProtoException("Not authenticated with Bluesky. Please connect first."); + } + String usedAccessJwt = current.getAccessJwt(); + try { + HttpRequest request = HttpRequest.newBuilder(context.getURIForLexicon(lexicon, params)) + .GET() + .timeout(REQUEST_TIMEOUT) + .header("Authorization", "Bearer " + usedAccessJwt) + .build(); + HttpResponse response = client.send(request, HttpResponse.BodyHandlers.ofString()); + int statusCode = response.statusCode(); + if (statusCode == 200) { + return objectMapper.readValue(response.body(), type); + } + if (statusCode == 401 && !refreshed) { + refreshed = true; + if (refreshSession(usedAccessJwt)) { + continue; // retry immediately with the new token (not counted as a retry) + } + throw new AtProtoException(statusCode, lexicon, response.body()); } - params.put("cursor", objectResponse.getCursor()); - currentCrawlLoop++; + if (statusCode == 429 && attempt < MAX_RETRIES) { + // Rate limited: wait it out and surface the countdown to the caller. + backoff(response, attempt++, onWaiting); + continue; + } + if (statusCode >= 500 && attempt < MAX_RETRIES) { + backoff(response, attempt++, null); + continue; + } + throw new AtProtoException(statusCode, lexicon, response.body()); + } catch (IOException e) { + if (attempt < MAX_RETRIES) { + backoff(null, attempt++, null); + continue; + } + throw new AtProtoException("XRPC " + lexicon + " failed after retries", e); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AtProtoException("XRPC " + lexicon + " interrupted", e); } - return pagedResponse; - } catch (IOException | InterruptedException e) { - throw new RuntimeException(e); } } - public AppBskyActorGetProfile appBskyActorGetProfile(String actor) { - try { - var params = new HashMap(); - params.put("actor", actor); - var request = getRequest("app.bsky.actor.getProfile", params); - var response = client.send(request, HttpResponse.BodyHandlers.ofString()); - System.out.println(response.body()); - return objectMapper.readValue(response.body(), AppBskyActorGetProfile.class); - } catch (IOException | InterruptedException ex) { - Exceptions.printStackTrace(ex); + /** + * Follows an XRPC cursor, handing each page to {@code onPage} as it arrives + * so callers can write results incrementally (and be cancelled between + * pages) instead of buffering an entire crawl in memory. + */ + private void paginate(String lexicon, Map baseParams, + Class type, Optional maxPages, Consumer onPage, LongConsumer onWaiting) { + Map params = new HashMap<>(baseParams); + int page = 0; + while (maxPages.isEmpty() || page < maxPages.get()) { + if (Thread.currentThread().isInterrupted()) { + return; + } + T response = get(lexicon, params, type, onWaiting); + onPage.accept(response); + if (response.getCursor() == null) { + return; + } + params.put("cursor", response.getCursor()); + page++; } - return null; } - - public List appBskyGraphGetList(String list) { - List lists = new ArrayList<>(); + + /** + * Sleeps before a retry. On a 429 it honors the {@code RateLimit-Reset} + * (epoch seconds) or {@code Retry-After} header; otherwise it uses + * randomized exponential backoff. Responds to interruption for cancellation. + */ + private void backoff(HttpResponse response, int attempt, LongConsumer onWaiting) { + long waitMs = Math.min(computeWaitMs(response, attempt) + ThreadLocalRandom.current().nextLong(250), MAX_BACKOFF_MS); + long deadline = System.currentTimeMillis() + waitMs; try { - var params = new HashMap(); - params.put("list", list); - params.put("limit", "100"); - while (true) { - var request = getRequest("app.bsky.graph.getList", params); - var response = client.send(request, HttpResponse.BodyHandlers.ofString()); - var objectResponse = objectMapper.readValue(response.body(), AppBskyGraphGetList.class); - lists.add(objectResponse); - System.out.println(response.body()); - if (objectResponse.getCursor() == null) { - break; + long remaining; + while ((remaining = deadline - System.currentTimeMillis()) > 0) { + if (onWaiting != null) { + onWaiting.accept(remaining); } - - params.put("cursor", objectResponse.getCursor()); + // Tick ~once per second so the caller can render a live countdown, + // and so cancellation (interrupt) is picked up promptly. + Thread.sleep(Math.min(remaining, 1000L)); + } + if (onWaiting != null) { + onWaiting.accept(0L); } - } catch (IOException | InterruptedException ex) { - Exceptions.printStackTrace(ex); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AtProtoException("Interrupted during backoff", e); } - return lists; } - public AppBskyActorGetProfile appBskyActorGetProfiles(String actors) { - try { - var request = HttpRequest - .newBuilder(context.getURIForLexicon("app.bsky.actor.getProfiles", actors)) - .GET() - .header("Authorization", "Bearer " + session.getAccessJwt()) - .build(); - var response = client.send(request, HttpResponse.BodyHandlers.ofString()); - return objectMapper.readValue(response.body(), AppBskyActorGetProfile.class); - } catch (IOException | InterruptedException ex) { - Exceptions.printStackTrace(ex); + private long computeWaitMs(HttpResponse response, int attempt) { + long waitMs = (long) (Math.pow(2, attempt) * 500L); // 0.5s, 1s, 2s, 4s, 8s, ... + if (response != null) { + Optional reset = response.headers().firstValue("ratelimit-reset"); + Optional retryAfter = response.headers().firstValue("retry-after"); + if (reset.isPresent()) { + try { + long resetMs = Long.parseLong(reset.get().trim()) * 1000L - System.currentTimeMillis(); + if (resetMs > 0) { + waitMs = resetMs; + } + } catch (NumberFormatException ignored) { + // fall back to exponential backoff + } + } else if (retryAfter.isPresent()) { + try { + long retryMs = Long.parseLong(retryAfter.get().trim()) * 1000L; + if (retryMs > 0) { + waitMs = retryMs; + } + } catch (NumberFormatException ignored) { + // fall back to exponential backoff + } + } } - return null; + return waitMs; + } + + public void appBskyGraphGetFollowers(String actor, Optional maxPages, Consumer onPage, LongConsumer onWaiting) { + paginate("app.bsky.graph.getFollowers", Map.of("actor", actor, "limit", "100"), + AppBskyGraphGetFollowers.class, maxPages, onPage, onWaiting); + } + + public void appBskyGraphGetFollows(String actor, Optional maxPages, Consumer onPage, LongConsumer onWaiting) { + paginate("app.bsky.graph.getFollows", Map.of("actor", actor, "limit", "100"), + AppBskyGraphGetFollows.class, maxPages, onPage, onWaiting); + } + + public void appBskyGraphGetList(String list, Consumer onPage, LongConsumer onWaiting) { + paginate("app.bsky.graph.getList", Map.of("list", list, "limit", "100"), + AppBskyGraphGetList.class, Optional.empty(), onPage, onWaiting); } } diff --git a/modules/BlueskyGephi/src/main/java/fr/totetmatt/blueskygephi/atproto/AtContext.java b/modules/BlueskyGephi/src/main/java/fr/totetmatt/blueskygephi/atproto/AtContext.java index 42c70824a..1f3ae89eb 100644 --- a/modules/BlueskyGephi/src/main/java/fr/totetmatt/blueskygephi/atproto/AtContext.java +++ b/modules/BlueskyGephi/src/main/java/fr/totetmatt/blueskygephi/atproto/AtContext.java @@ -7,7 +7,7 @@ import java.net.URI; import java.net.URLEncoder; import java.nio.charset.StandardCharsets; -import java.util.HashMap; +import java.util.Map; import java.util.stream.Collectors; public class AtContext { @@ -27,7 +27,7 @@ public URI getURIForLexicon(String lexicon) { return URI.create(formatUrl(host, lexicon)); } - public URI getURIForLexicon(String lexicon, HashMap parameters) { + public URI getURIForLexicon(String lexicon, Map parameters) { String url_parameters = parameters.entrySet().stream().map(x -> URLEncoder.encode(x.getKey(), StandardCharsets.UTF_8) + "=" + URLEncoder.encode(x.getValue(), StandardCharsets.UTF_8) @@ -36,9 +36,4 @@ public URI getURIForLexicon(String lexicon, HashMap parameters) return URI.create(formatUrl(host, lexicon) + "?" + url_parameters); } - public URI getURIForLexicon(String lexicon, String parameters) { - - return URI.create(formatUrl(host, lexicon) + "?" + parameters); - } - } diff --git a/modules/BlueskyGephi/src/main/java/fr/totetmatt/blueskygephi/atproto/AtProtoException.java b/modules/BlueskyGephi/src/main/java/fr/totetmatt/blueskygephi/atproto/AtProtoException.java new file mode 100644 index 000000000..ed3bb1fc9 --- /dev/null +++ b/modules/BlueskyGephi/src/main/java/fr/totetmatt/blueskygephi/atproto/AtProtoException.java @@ -0,0 +1,33 @@ +package fr.totetmatt.blueskygephi.atproto; + +/** + * Uniform error type for all XRPC calls, so callers get one consistent failure + * mode instead of a mix of thrown {@link RuntimeException}s and {@code null} + * returns. + * + * @author totetmatt + */ +public class AtProtoException extends RuntimeException { + + /** HTTP status when the failure came from a response, otherwise -1. */ + private final int statusCode; + + public AtProtoException(String message) { + super(message); + this.statusCode = -1; + } + + public AtProtoException(String message, Throwable cause) { + super(message, cause); + this.statusCode = -1; + } + + public AtProtoException(int statusCode, String lexicon, String body) { + super("XRPC " + lexicon + " failed (HTTP " + statusCode + "): " + body); + this.statusCode = statusCode; + } + + public int getStatusCode() { + return statusCode; + } +} diff --git a/modules/BlueskyGephi/src/main/java/fr/totetmatt/blueskygephi/atproto/response/AppBskyGraphGetFollowers.java b/modules/BlueskyGephi/src/main/java/fr/totetmatt/blueskygephi/atproto/response/AppBskyGraphGetFollowers.java index 75e095381..78b065979 100644 --- a/modules/BlueskyGephi/src/main/java/fr/totetmatt/blueskygephi/atproto/response/AppBskyGraphGetFollowers.java +++ b/modules/BlueskyGephi/src/main/java/fr/totetmatt/blueskygephi/atproto/response/AppBskyGraphGetFollowers.java @@ -7,7 +7,7 @@ import fr.totetmatt.blueskygephi.atproto.response.common.Identity; import java.util.List; -public class AppBskyGraphGetFollowers { +public class AppBskyGraphGetFollowers implements Paged { private Identity subject; diff --git a/modules/BlueskyGephi/src/main/java/fr/totetmatt/blueskygephi/atproto/response/AppBskyGraphGetFollows.java b/modules/BlueskyGephi/src/main/java/fr/totetmatt/blueskygephi/atproto/response/AppBskyGraphGetFollows.java index 950fba017..f956e2e09 100644 --- a/modules/BlueskyGephi/src/main/java/fr/totetmatt/blueskygephi/atproto/response/AppBskyGraphGetFollows.java +++ b/modules/BlueskyGephi/src/main/java/fr/totetmatt/blueskygephi/atproto/response/AppBskyGraphGetFollows.java @@ -7,7 +7,7 @@ import fr.totetmatt.blueskygephi.atproto.response.common.Identity; import java.util.List; -public class AppBskyGraphGetFollows { +public class AppBskyGraphGetFollows implements Paged { private Identity subject; diff --git a/modules/BlueskyGephi/src/main/java/fr/totetmatt/blueskygephi/atproto/response/AppBskyGraphGetList.java b/modules/BlueskyGephi/src/main/java/fr/totetmatt/blueskygephi/atproto/response/AppBskyGraphGetList.java index 7702d8291..af850d124 100644 --- a/modules/BlueskyGephi/src/main/java/fr/totetmatt/blueskygephi/atproto/response/AppBskyGraphGetList.java +++ b/modules/BlueskyGephi/src/main/java/fr/totetmatt/blueskygephi/atproto/response/AppBskyGraphGetList.java @@ -11,7 +11,7 @@ * * @author totetmatt */ -public class AppBskyGraphGetList { +public class AppBskyGraphGetList implements Paged { private List items; private String cursor; diff --git a/modules/BlueskyGephi/src/main/java/fr/totetmatt/blueskygephi/atproto/response/Paged.java b/modules/BlueskyGephi/src/main/java/fr/totetmatt/blueskygephi/atproto/response/Paged.java new file mode 100644 index 000000000..c0fecb990 --- /dev/null +++ b/modules/BlueskyGephi/src/main/java/fr/totetmatt/blueskygephi/atproto/response/Paged.java @@ -0,0 +1,16 @@ +package fr.totetmatt.blueskygephi.atproto.response; + +/** + * A cursor-paginated XRPC response. Implemented by every response type that the + * generic pagination helper in {@code AtClient} can walk. + * + * @author totetmatt + */ +public interface Paged { + + /** + * @return the cursor for the next page, or {@code null} when there are no + * more pages. + */ + String getCursor(); +}