✨ (federation): Show received pushes on the Following page
Changes
11 files changed, +371 -5
MODIFY
README.md
+5 -2
@@ -52,8 +52,11 @@
52
52
- On push, a `Push` activity is published to the outbox and delivered to followers via a persisted,
53
53
retrying, HTTP-Signature-signed delivery queue
54
54
- Outbound: logged-in users can follow a remote repository/actor from the **Following** page by
55
- handle (`owner/name@host`) or actor URL; the handle is resolved via WebFinger and a signed `Follow`
56
- is enqueued for delivery, tracked as PENDING until the remote `Accept`s (unfollow sends `Undo`)
55
+ handle (`owner/name@host`) or actor URL; the handle is resolved via WebFinger (lenient about
56
+ bare-host vs. host:port `acct:` forms) and a signed `Follow` is enqueued for delivery, tracked as
57
+ PENDING until the remote `Accept`s (unfollow sends `Undo`)
58
+ - Inbound `Push` activities from followed repositories are stored and surfaced as a "Recent
59
+ pushes" feed on the Following page; pushes from actors nobody follows are dropped
57
60
58
61
Inbound activities must carry a valid HTTP Signature from an **allowlisted** peer; outbound fetches
59
62
are HTTPS-only, allowlist-bound, and SSRF-guarded (no private/loopback/link-local targets). Issues
MODIFY
src/main/java/de/workaround/federation/ActivityDispatcher.java
+4 -0
@@ -25,6 +25,9 @@
25
25
@Inject
26
26
AcceptHandler acceptHandler;
27
27
28
+ @Inject
29
+ PushHandler pushHandler;
30
+
28
31
/** Called within the inbox receipt transaction, after the activity id has been recorded. */
29
32
public void dispatch(JsonNode activity)
30
33
{
@@ -34,6 +37,7 @@
34
37
case "Follow" -> followHandler.handle(activity);
35
38
case "Undo" -> undoHandler.handle(activity);
36
39
case "Accept" -> acceptHandler.handle(activity);
40
+ case "Push" -> pushHandler.handle(activity);
37
41
default -> LOG.debugf("Ignoring unsupported activity type: %s", type);
38
42
}
39
43
}
ADD
src/main/java/de/workaround/federation/PushHandler.java
+50 -0
@@ -0,0 +1,50 @@
1
+package de.workaround.federation;
2
+
3
+import com.fasterxml.jackson.databind.JsonNode;
4
+
5
+import de.workaround.model.ReceivedPush;
6
+import de.workaround.model.RemoteFollow;
7
+import jakarta.enterprise.context.ApplicationScoped;
8
+import jakarta.inject.Inject;
9
+import jakarta.transaction.Transactional;
10
+
11
+/**
12
+ * Handles an inbound {@code Push} from a remote repository: stored for the follower-side feed, but
13
+ * only while at least one local user follows the sending actor — anything else is dropped.
14
+ */
15
+@ApplicationScoped
16
+public class PushHandler
17
+{
18
+ @Inject
19
+ RemoteFollow.Repo follows;
20
+
21
+ @Inject
22
+ ReceivedPush.Repo pushes;
23
+
24
+ @Transactional
25
+ public void handle(JsonNode push)
26
+ {
27
+ String actorId = LocalActors.idOf(push.path("actor"));
28
+ String activityId = push.path("id").asText(null);
29
+ if (actorId == null || activityId == null)
30
+ {
31
+ return;
32
+ }
33
+ if (follows.findByRemoteActorId(actorId).isEmpty())
34
+ {
35
+ return; // nobody follows this repository — not our feed
36
+ }
37
+ if (pushes.findByActivityId(activityId).isPresent())
38
+ {
39
+ return; // redelivered — already in the feed
40
+ }
41
+ ReceivedPush received = new ReceivedPush();
42
+ received.remoteActorId = actorId;
43
+ received.activityId = activityId;
44
+ received.summary = push.path("summary").asText(null);
45
+ received.target = push.path("target").asText(null);
46
+ received.payload = push.toString();
47
+ received.persist();
48
+ }
49
+
50
+}
MODIFY
src/main/java/de/workaround/federation/RemoteFollowService.java
+19 -0
@@ -8,6 +8,7 @@
8
8
import com.fasterxml.jackson.databind.node.ObjectNode;
9
9
10
10
import de.workaround.model.FederationKey;
11
+import de.workaround.model.ReceivedPush;
11
12
import de.workaround.model.RemoteActor;
12
13
import de.workaround.model.RemoteFollow;
13
14
import de.workaround.model.User;
@@ -51,10 +52,15 @@
51
52
@Inject
52
53
DeliveryService delivery;
53
54
55
+ private static final int FEED_LIMIT = 50;
56
+
54
57
@Inject
55
58
RemoteFollow.Repo follows;
56
59
57
60
@Inject
61
+ ReceivedPush.Repo pushes;
62
+
63
+ @Inject
58
64
User.Repo users;
59
65
60
66
@Inject
@@ -123,6 +129,19 @@
123
129
return follows.findByUser(user);
124
130
}
125
131
132
+ /** Recent {@code Push} activities of the repositories the user follows, newest first. */
133
+ @Transactional
134
+ public List<ReceivedPush> recentPushes(User user)
135
+ {
136
+ List<String> actorIds = follows.findByUser(user).stream().map(f -> f.remoteActorId).toList();
137
+ if (actorIds.isEmpty())
138
+ {
139
+ return List.of();
140
+ }
141
+ List<ReceivedPush> feed = pushes.findByRemoteActorIds(actorIds);
142
+ return feed.size() > FEED_LIMIT ? feed.subList(0, FEED_LIMIT) : feed;
143
+ }
144
+
126
145
/** Resolves the input to a remote actor id: pass URLs through, WebFinger-resolve handles. */
127
146
private String resolveActorId(String input)
128
147
{
ADD
src/main/java/de/workaround/model/ReceivedPush.java
+59 -0
@@ -0,0 +1,59 @@
1
+package de.workaround.model;
2
+
3
+import java.time.Instant;
4
+import java.util.List;
5
+import java.util.Optional;
6
+import java.util.UUID;
7
+
8
+import org.hibernate.annotations.processing.Find;
9
+import org.hibernate.annotations.processing.HQL;
10
+
11
+import io.quarkus.hibernate.panache.PanacheEntity;
12
+import io.quarkus.hibernate.panache.PanacheRepository;
13
+import jakarta.persistence.Column;
14
+import jakarta.persistence.Entity;
15
+import jakarta.persistence.GeneratedValue;
16
+import jakarta.persistence.GenerationType;
17
+import jakarta.persistence.Id;
18
+import jakarta.persistence.Table;
19
+
20
+/**
21
+ * A {@code Push} activity received from a followed remote repository, kept so following users can
22
+ * see what happened in repositories they follow. {@code remoteActorId} is the sending repository's
23
+ * actor id; {@code activityId} deduplicates redeliveries.
24
+ */
25
+@Entity
26
+@Table(name = "received_pushes")
27
+public class ReceivedPush implements PanacheEntity.Managed
28
+{
29
+ @Id
30
+ @GeneratedValue(strategy = GenerationType.UUID)
31
+ public UUID id;
32
+
33
+ @Column(columnDefinition = "text")
34
+ public String remoteActorId;
35
+
36
+ @Column(columnDefinition = "text")
37
+ public String activityId;
38
+
39
+ @Column(columnDefinition = "text")
40
+ public String summary;
41
+
42
+ @Column(columnDefinition = "text")
43
+ public String target;
44
+
45
+ @Column(columnDefinition = "text")
46
+ public String payload;
47
+
48
+ public Instant receivedAt = Instant.now();
49
+
50
+ public interface Repo extends PanacheRepository.Managed<ReceivedPush, UUID>
51
+ {
52
+ @Find
53
+ Optional<ReceivedPush> findByActivityId(String activityId);
54
+
55
+ @HQL("select p from ReceivedPush p where p.remoteActorId in :remoteActorIds order by p.receivedAt desc")
56
+ List<ReceivedPush> findByRemoteActorIds(List<String> remoteActorIds);
57
+ }
58
+
59
+}
MODIFY
src/main/java/de/workaround/model/RemoteFollow.java
+3 -0
@@ -67,6 +67,9 @@
67
67
@Find
68
68
Optional<RemoteFollow> findByFollowActivityId(String followActivityId);
69
69
70
+ @Find
71
+ List<RemoteFollow> findByRemoteActorId(String remoteActorId);
72
+
70
73
@HQL("select f from RemoteFollow f where f.user = :user order by f.createdAt")
71
74
List<RemoteFollow> findByUser(User user);
72
75
}
MODIFY
src/main/java/de/workaround/web/FollowingResource.java
+6 -3
@@ -6,6 +6,7 @@
6
6
7
7
import de.workaround.account.CurrentUser;
8
8
import de.workaround.federation.RemoteFollowService;
9
+import de.workaround.model.ReceivedPush;
9
10
import de.workaround.model.RemoteFollow;
10
11
import de.workaround.model.User;
11
12
import io.quarkus.qute.CheckedTemplate;
@@ -29,7 +30,8 @@
29
30
@CheckedTemplate
30
31
static class Templates
31
32
{
32
- static native TemplateInstance following(List<RemoteFollow> follows, String error);
33
+ static native TemplateInstance following(List<RemoteFollow> follows, List<ReceivedPush> pushes,
34
+ String error);
33
35
}
34
36
35
37
@Inject
@@ -41,7 +43,8 @@
41
43
@GET
42
44
public TemplateInstance list()
43
45
{
44
- return Templates.following(service.list(currentUser.require()), null);
46
+ User user = currentUser.require();
47
+ return Templates.following(service.list(user), service.recentPushes(user), null);
45
48
}
46
49
47
50
@POST
@@ -57,7 +60,7 @@
57
60
catch (RemoteFollowService.RemoteFollowException e)
58
61
{
59
62
return Response.status(Response.Status.BAD_REQUEST)
60
- .entity(Templates.following(service.list(user), e.getMessage()))
63
+ .entity(Templates.following(service.list(user), service.recentPushes(user), e.getMessage()))
61
64
.build();
62
65
}
63
66
}
ADD
src/main/resources/db/migration/V10__received_pushes.sql
+15 -0
@@ -0,0 +1,15 @@
1
+-- Follower-side feed: Push activities received from followed remote repositories.
2
+-- remote_actor_id is the sending repository's actor id; rows are only stored while
3
+-- at least one local user follows that actor.
4
+create table received_pushes
5
+(
6
+ id uuid primary key,
7
+ remote_actor_id text not null,
8
+ activity_id text not null unique,
9
+ summary text,
10
+ target text,
11
+ payload text not null,
12
+ received_at timestamptz not null default now()
13
+);
14
+
15
+create index idx_received_pushes_actor on received_pushes (remote_actor_id, received_at);
MODIFY
src/main/resources/templates/FollowingResource/following.html
+14 -0
@@ -19,6 +19,20 @@
19
19
</tr>
20
20
{/for}
21
21
</table>
22
+{#if pushes}
23
+<h2>Recent pushes</h2>
24
+<table>
25
+ <tr><th>Repository</th><th>Ref</th><th>Summary</th><th>Received</th></tr>
26
+ {#for push in pushes}
27
+ <tr>
28
+ <td><a href="{push.remoteActorId}" rel="noopener noreferrer">{push.remoteActorId}</a></td>
29
+ <td>{push.target}</td>
30
+ <td>{push.summary}</td>
31
+ <td>{push.receivedAt}</td>
32
+ </tr>
33
+ {/for}
34
+</table>
35
+{/if}
22
36
<h2>Follow remote repository</h2>
23
37
<form method="post" action="/following">
24
38
<p><label>Handle or actor URL <input name="handle" placeholder="owner/name@remote-host" required></label></p>
ADD
src/test/java/de/workaround/federation/PushFeedTest.java
+149 -0
@@ -0,0 +1,149 @@
1
+package de.workaround.federation;
2
+
3
+import java.time.Instant;
4
+import java.util.List;
5
+import java.util.UUID;
6
+
7
+import org.junit.jupiter.api.Test;
8
+
9
+import com.fasterxml.jackson.databind.ObjectMapper;
10
+import com.fasterxml.jackson.databind.node.ObjectNode;
11
+
12
+import de.workaround.model.ReceivedPush;
13
+import de.workaround.model.RemoteActor;
14
+import de.workaround.model.User;
15
+import io.quarkus.test.junit.QuarkusTest;
16
+import jakarta.inject.Inject;
17
+import jakarta.transaction.Transactional;
18
+
19
+import static org.junit.jupiter.api.Assertions.assertEquals;
20
+import static org.junit.jupiter.api.Assertions.assertTrue;
21
+
22
+/**
23
+ * Follower-side feed of received {@code Push} activities: pushes from followed repositories are
24
+ * stored and listed for the following user; pushes from unfollowed actors are ignored.
25
+ */
26
+@QuarkusTest
27
+class PushFeedTest
28
+{
29
+ @Inject
30
+ RemoteFollowService service;
31
+
32
+ @Inject
33
+ ActivityDispatcher dispatcher;
34
+
35
+ @Inject
36
+ ReceivedPush.Repo pushes;
37
+
38
+ @Inject
39
+ User.Repo userRepo;
40
+
41
+ @Inject
42
+ ObjectMapper mapper;
43
+
44
+ @Test
45
+ void pushFromFollowedRepoIsStoredAndListed()
46
+ {
47
+ User user = persistUser("pf-alice-" + unique());
48
+ String remote = "https://peer.test/ap/repos/bob/lib-" + unique();
49
+ seedRemoteActor(remote, remote + "/inbox");
50
+ service.follow(user, remote);
51
+
52
+ dispatcher.dispatch(push(remote, "Pushed 2 commit(s) to refs/heads/main"));
53
+
54
+ List<ReceivedPush> feed = service.recentPushes(user);
55
+ assertEquals(1, feed.size());
56
+ assertEquals(remote, feed.get(0).remoteActorId);
57
+ assertEquals("Pushed 2 commit(s) to refs/heads/main", feed.get(0).summary);
58
+ assertEquals("refs/heads/main", feed.get(0).target);
59
+ }
60
+
61
+ @Test
62
+ void pushFromUnfollowedActorIsIgnored()
63
+ {
64
+ User user = persistUser("pf-bob-" + unique());
65
+ String stranger = "https://peer.test/ap/repos/mallory/junk-" + unique();
66
+
67
+ dispatcher.dispatch(push(stranger, "Pushed 1 commit(s) to refs/heads/main"));
68
+
69
+ assertTrue(service.recentPushes(user).isEmpty());
70
+ assertTrue(findByActor(stranger).isEmpty());
71
+ }
72
+
73
+ @Test
74
+ void duplicatePushActivityIsStoredOnce()
75
+ {
76
+ User user = persistUser("pf-carol-" + unique());
77
+ String remote = "https://peer.test/ap/repos/bob/lib-" + unique();
78
+ seedRemoteActor(remote, remote + "/inbox");
79
+ service.follow(user, remote);
80
+
81
+ ObjectNode push = push(remote, "Pushed 1 commit(s) to refs/heads/main");
82
+ dispatcher.dispatch(push);
83
+ dispatcher.dispatch(push);
84
+
85
+ assertEquals(1, service.recentPushes(user).size());
86
+ }
87
+
88
+ @Test
89
+ void feedOnlyShowsPushesOfOwnFollows()
90
+ {
91
+ User follower = persistUser("pf-dan-" + unique());
92
+ User other = persistUser("pf-erin-" + unique());
93
+ String remote = "https://peer.test/ap/repos/bob/lib-" + unique();
94
+ seedRemoteActor(remote, remote + "/inbox");
95
+ service.follow(follower, remote);
96
+
97
+ dispatcher.dispatch(push(remote, "Pushed 1 commit(s) to refs/heads/main"));
98
+
99
+ assertEquals(1, service.recentPushes(follower).size());
100
+ assertTrue(service.recentPushes(other).isEmpty());
101
+ }
102
+
103
+ private ObjectNode push(String actor, String summary)
104
+ {
105
+ ObjectNode node = mapper.createObjectNode();
106
+ node.put("id", actor + "/activities/" + UUID.randomUUID());
107
+ node.put("type", "Push");
108
+ node.put("actor", actor);
109
+ node.put("context", actor);
110
+ node.put("target", "refs/heads/main");
111
+ node.put("summary", summary);
112
+ return node;
113
+ }
114
+
115
+ @Transactional
116
+ List<ReceivedPush> findByActor(String actorId)
117
+ {
118
+ return pushes.findByRemoteActorIds(List.of(actorId));
119
+ }
120
+
121
+ @Transactional
122
+ void seedRemoteActor(String actorId, String inbox)
123
+ {
124
+ RemoteActor actor = new RemoteActor();
125
+ actor.actorId = actorId;
126
+ actor.inbox = inbox;
127
+ actor.publicKeyPem = "-----BEGIN PUBLIC KEY-----\nstub\n-----END PUBLIC KEY-----";
128
+ actor.fetchedAt = Instant.now();
129
+ actor.persist();
130
+ }
131
+
132
+ @Transactional
133
+ User persistUser(String name)
134
+ {
135
+ return userRepo.findByOidcSubOptional(name).orElseGet(() -> {
136
+ User user = new User();
137
+ user.oidcSub = name;
138
+ user.username = name;
139
+ user.persist();
140
+ return user;
141
+ });
142
+ }
143
+
144
+ private static String unique()
145
+ {
146
+ return UUID.randomUUID().toString().substring(0, 8);
147
+ }
148
+
149
+}
MODIFY
src/test/java/de/workaround/web/FollowingPagesTest.java
+47 -0
@@ -5,9 +5,14 @@
5
5
6
6
import org.junit.jupiter.api.Test;
7
7
8
+import com.fasterxml.jackson.databind.ObjectMapper;
9
+import com.fasterxml.jackson.databind.node.ObjectNode;
10
+
11
+import de.workaround.federation.ActivityDispatcher;
8
12
import de.workaround.model.RemoteActor;
9
13
import io.quarkus.test.junit.QuarkusTest;
10
14
import io.quarkus.test.security.TestSecurity;
15
+import jakarta.inject.Inject;
11
16
import jakarta.transaction.Transactional;
12
17
13
18
import static io.restassured.RestAssured.given;
@@ -20,6 +25,12 @@
20
25
@QuarkusTest
21
26
class FollowingPagesTest
22
27
{
28
+ @Inject
29
+ ActivityDispatcher dispatcher;
30
+
31
+ @Inject
32
+ ObjectMapper mapper;
33
+
23
34
@Test
24
35
@TestSecurity(user = "following-tester")
25
36
void followListAndUnfollowRoundTrip()
@@ -65,6 +76,30 @@
65
76
}
66
77
67
78
@Test
79
+ @TestSecurity(user = "push-feed-tester")
80
+ void receivedPushesFromFollowedRepoAppearOnPage()
81
+ {
82
+ String remote = "https://peer.test/ap/repos/bob/feed-" + unique();
83
+ seedRemoteActor(remote, remote + "/inbox");
84
+
85
+ given()
86
+ .redirects().follow(false)
87
+ .formParam("handle", remote)
88
+ .when().post("/following")
89
+ .then()
90
+ .statusCode(anyOf(is(302), is(303)));
91
+
92
+ dispatchPush(remote, "Pushed 3 commit(s) to refs/heads/main");
93
+
94
+ given()
95
+ .when().get("/following")
96
+ .then()
97
+ .statusCode(200)
98
+ .body(containsString("Recent pushes"))
99
+ .body(containsString("Pushed 3 commit(s) to refs/heads/main"));
100
+ }
101
+
102
+ @Test
68
103
@TestSecurity(user = "following-err-tester")
69
104
void unresolvableHandleShowsError()
70
105
{
@@ -87,6 +122,18 @@
87
122
}
88
123
89
124
@Transactional
125
+ void dispatchPush(String actor, String summary)
126
+ {
127
+ ObjectNode node = mapper.createObjectNode();
128
+ node.put("id", actor + "/activities/" + UUID.randomUUID());
129
+ node.put("type", "Push");
130
+ node.put("actor", actor);
131
+ node.put("target", "refs/heads/main");
132
+ node.put("summary", summary);
133
+ dispatcher.dispatch(node);
134
+ }
135
+
136
+ @Transactional
90
137
void seedRemoteActor(String actorId, String inbox)
91
138
{
92
139
RemoteActor actor = new RemoteActor();