Skip to content

Commit 20e7672

Browse files
tranktenroot
trankten
and
root
authored
feat: Add ActivityPub Relay support (#352)
* feat: implement ActivityPub relay support - Add relay subscription management system - Create RelaySubscription model and migration - Implement RelayService for managing relay subscriptions - Add automatic delivery to relays for public posts - Process incoming activities from subscribed relays - Create admin panel for relay management - Add relay statistics and monitoring * fix: Correct relay delivery integration and visibility check --------- Co-authored-by: root <root@hydra.tkz.local>
1 parent b5229a6 commit 20e7672

12 files changed

Lines changed: 903 additions & 25 deletions

File tree

‎app/Http/Controllers/AdminRelayController.php‎

Lines changed: 149 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,149 @@
1+
<?php
2+
3+
namespace App\Http\Controllers;
4+
5+
use App\Http\Middleware\AdminOnlyAccess;
6+
use App\Models\RelaySubscription;
7+
use App\Services\RelayService;
8+
use Illuminate\Http\JsonResponse;
9+
use Illuminate\Http\Request;
10+
use Illuminate\Support\Facades\Log;
11+
use Exception;
12+
13+
class AdminRelayController extends Controller
14+
{
15+
protected $relayService;
16+
17+
public function __construct(RelayService $relayService)
18+
{
19+
$this->middleware(AdminOnlyAccess::class);
20+
$this->relayService = $relayService;
21+
}
22+
23+
public function index(): JsonResponse
24+
{
25+
$relays = RelaySubscription::orderBy('created_at', 'desc')->get();
26+
27+
return response()->json([
28+
'relays' => $relays,
29+
]);
30+
}
31+
32+
public function store(Request $request): JsonResponse
33+
{
34+
$validated = $request->validate([
35+
'relay_url' => 'required|url|unique:relay_subscriptions,relay_url',
36+
'name' => 'nullable|string|max:255',
37+
'send_public_posts' => 'boolean',
38+
'receive_content' => 'boolean',
39+
]);
40+
41+
try {
42+
$relay = $this->relayService->subscribe($validated['relay_url']);
43+
44+
if (isset($validated['name'])) {
45+
$relay->update(['name' => $validated['name']]);
46+
}
47+
48+
if (isset($validated['send_public_posts'])) {
49+
$relay->update(['send_public_posts' => $validated['send_public_posts']]);
50+
}
51+
52+
if (isset($validated['receive_content'])) {
53+
$relay->update(['receive_content' => $validated['receive_content']]);
54+
}
55+
56+
return response()->json([
57+
'message' => 'Relay subscription initiated',
58+
'relay' => $relay,
59+
], 201);
60+
61+
} catch (Exception $e) {
62+
if (config('logging.dev_log')) {
63+
Log::error('Failed to subscribe to relay', [
64+
'relay_url' => $validated['relay_url'],
65+
'error' => $e->getMessage(),
66+
]);
67+
}
68+
69+
return response()->json([
70+
'error' => 'Failed to subscribe to relay',
71+
'message' => $e->getMessage(),
72+
], 500);
73+
}
74+
}
75+
76+
public function update(Request $request, RelaySubscription $relay): JsonResponse
77+
{
78+
$validated = $request->validate([
79+
'name' => 'nullable|string|max:255',
80+
'send_public_posts' => 'boolean',
81+
'receive_content' => 'boolean',
82+
'status' => 'in:active,disabled',
83+
]);
84+
85+
$relay->update($validated);
86+
87+
return response()->json([
88+
'message' => 'Relay updated successfully',
89+
'relay' => $relay->fresh(),
90+
]);
91+
}
92+
93+
public function destroy(RelaySubscription $relay): JsonResponse
94+
{
95+
try {
96+
$this->relayService->unsubscribe($relay);
97+
98+
return response()->json([
99+
'message' => 'Relay unsubscribed successfully',
100+
]);
101+
102+
} catch (Exception $e) {
103+
if (config('logging.dev_log')) {
104+
Log::error('Failed to unsubscribe from relay', [
105+
'relay_id' => $relay->id,
106+
'error' => $e->getMessage(),
107+
]);
108+
}
109+
110+
return response()->json([
111+
'error' => 'Failed to unsubscribe from relay',
112+
'message' => $e->getMessage(),
113+
], 500);
114+
}
115+
}
116+
117+
public function enable(RelaySubscription $relay): JsonResponse
118+
{
119+
$relay->enable();
120+
121+
return response()->json([
122+
'message' => 'Relay enabled successfully',
123+
'relay' => $relay->fresh(),
124+
]);
125+
}
126+
127+
public function disable(RelaySubscription $relay): JsonResponse
128+
{
129+
$relay->disable();
130+
131+
return response()->json([
132+
'message' => 'Relay disabled successfully',
133+
'relay' => $relay->fresh(),
134+
]);
135+
}
136+
137+
public function stats(): JsonResponse
138+
{
139+
$stats = [
140+
'total_relays' => RelaySubscription::count(),
141+
'active_relays' => RelaySubscription::where('status', 'active')->count(),
142+
'pending_relays' => RelaySubscription::where('status', 'pending')->count(),
143+
'total_sent' => RelaySubscription::sum('total_sent'),
144+
'total_received' => RelaySubscription::sum('total_received'),
145+
];
146+
147+
return response()->json(['stats' => $stats]);
148+
}
149+
}

‎app/Jobs/Federation/DeliverCreateVideoActivity.php‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -82,6 +82,10 @@ public function handle(): void
8282

8383
$activity = CreateActivityBuilder::buildForVideo($actor, $video);
8484

85+
if ($actor->local && $video->visibility == 1) {
86+
app(\App\Services\RelayService::class)->deliverToRelays($actor, $activity);
87+
}
88+
8589
$parsedUrl = $this->parsedUrl;
8690

8791
$headers = [

‎app/Jobs/Federation/ProcessSharedInboxActivity.php‎

Lines changed: 55 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -77,14 +77,21 @@ public function handle()
7777
}
7878

7979
try {
80-
// Check if Delete Activity first, it will have no recipients
80+
$actorUrl = $this->activity['actor'] ?? null;
81+
if ($actorUrl) {
82+
$relay = app(\App\Services\RelayService::class)->isFromRelay($actorUrl);
83+
if ($relay) {
84+
$this->handleRelayActivity($relay);
85+
return;
86+
}
87+
}
88+
8189
if (isset($this->activity['type']) && $this->activity['type'] === 'Delete') {
8290
$this->handleDeleteActivity();
8391

8492
return;
8593
}
8694

87-
// Handle Follow activities - they don't have to/cc, only actor and object
8895
if (isset($this->activity['type']) && $this->activity['type'] === 'Follow') {
8996
$this->handleFollowActivity();
9097

@@ -368,6 +375,52 @@ protected function handleDeleteActivity()
368375
}
369376
}
370377

378+
protected function handleRelayActivity($relay)
379+
{
380+
$type = $this->activity['type'] ?? null;
381+
382+
if ($type === 'Accept' && isset($this->activity['object']['type']) && $this->activity['object']['type'] === 'Follow') {
383+
app(\App\Services\RelayService::class)->handleAccept($relay, $this->activity);
384+
if (config('logging.dev_log')) {
385+
Log::info('Relay subscription accepted', ['relay' => $relay->relay_url]);
386+
}
387+
return;
388+
}
389+
390+
if ($type === 'Reject' && isset($this->activity['object']['type']) && $this->activity['object']['type'] === 'Follow') {
391+
app(\App\Services\RelayService::class)->handleReject($relay);
392+
if (config('logging.dev_log')) {
393+
Log::warning('Relay subscription rejected', ['relay' => $relay->relay_url]);
394+
}
395+
return;
396+
}
397+
398+
if (in_array($type, ['Create', 'Announce', 'Update'])) {
399+
$recipients = $this->extractRecipients();
400+
$localTargets = $this->findLocalTargets($recipients);
401+
402+
if ($localTargets->isNotEmpty()) {
403+
$chunkSize = config('loops.federation.inbox_dispatch_chunk_size', 100);
404+
$localTargets->chunk($chunkSize)->each(function ($chunk) {
405+
foreach ($chunk as $target) {
406+
ProcessInboxActivity::dispatch($this->activity, $this->actor, $target)
407+
->onQueue('activitypub-in');
408+
}
409+
});
410+
}
411+
412+
$relay->incrementReceived();
413+
414+
if (config('logging.dev_log')) {
415+
Log::info('Processed relay activity', [
416+
'relay' => $relay->relay_url,
417+
'type' => $type,
418+
'targets' => $localTargets->count() ?? 0,
419+
]);
420+
}
421+
}
422+
}
423+
371424
/**
372425
* Handle a job failure.
373426
*/

‎app/Models/RelaySubscription.php‎

Lines changed: 91 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,91 @@
1+
<?php
2+
3+
namespace App\Models;
4+
5+
use Illuminate\Database\Eloquent\Model;
6+
7+
class RelaySubscription extends Model
8+
{
9+
protected $fillable = [
10+
'name',
11+
'relay_url',
12+
'relay_actor_url',
13+
'inbox_url',
14+
'shared_inbox_url',
15+
'status',
16+
'send_public_posts',
17+
'receive_content',
18+
'last_delivery_at',
19+
'last_received_at',
20+
'total_sent',
21+
'total_received',
22+
'relay_info',
23+
];
24+
25+
protected $casts = [
26+
'send_public_posts' => 'boolean',
27+
'receive_content' => 'boolean',
28+
'last_delivery_at' => 'datetime',
29+
'last_received_at' => 'datetime',
30+
'total_sent' => 'integer',
31+
'total_received' => 'integer',
32+
'relay_info' => 'array',
33+
];
34+
35+
public function isActive(): bool
36+
{
37+
return $this->status === 'active';
38+
}
39+
40+
public function isPending(): bool
41+
{
42+
return $this->status === 'pending';
43+
}
44+
45+
public function canSend(): bool
46+
{
47+
return $this->isActive() && $this->send_public_posts;
48+
}
49+
50+
public function canReceive(): bool
51+
{
52+
return $this->isActive() && $this->receive_content;
53+
}
54+
55+
public function getInbox(): ?string
56+
{
57+
return $this->shared_inbox_url ?? $this->inbox_url;
58+
}
59+
60+
public function incrementSent(): void
61+
{
62+
$this->increment('total_sent');
63+
$this->update(['last_delivery_at' => now()]);
64+
}
65+
66+
public function incrementReceived(): void
67+
{
68+
$this->increment('total_received');
69+
$this->update(['last_received_at' => now()]);
70+
}
71+
72+
public function markAsActive(): void
73+
{
74+
$this->update(['status' => 'active']);
75+
}
76+
77+
public function markAsRejected(): void
78+
{
79+
$this->update(['status' => 'rejected']);
80+
}
81+
82+
public function disable(): void
83+
{
84+
$this->update(['status' => 'disabled']);
85+
}
86+
87+
public function enable(): void
88+
{
89+
$this->update(['status' => 'active']);
90+
}
91+
}

‎app/Services/FederationDispatcher.php‎

Lines changed: 23 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -74,30 +74,32 @@ function ($inboxes) use (&$allInboxes) {
7474
'video_id' => $video->id,
7575
]);
7676
}
77+
} else {
78+
$jobs = $allInboxes->map(function ($inbox) use ($video) {
79+
return new DeliverCreateVideoActivity(
80+
$video,
81+
$inbox['inbox'],
82+
$inbox['profile_ids']
83+
);
84+
})->toArray();
85+
86+
Bus::batch($jobs)
87+
->name("Deliver Video {$video->id}")
88+
->allowFailures()
89+
->onQueue('activitypub-out')
90+
->dispatch();
7791

78-
return;
92+
if (config('logging.dev_log')) {
93+
Log::info('Video creation dispatched', [
94+
'profile_id' => $actor->id,
95+
'video_id' => $video->id,
96+
'inbox_count' => count($jobs),
97+
]);
98+
}
7999
}
80100

81-
$jobs = $allInboxes->map(function ($inbox) use ($video) {
82-
return new DeliverCreateVideoActivity(
83-
$video,
84-
$inbox['inbox'],
85-
$inbox['profile_ids']
86-
);
87-
})->toArray();
88-
89-
Bus::batch($jobs)
90-
->name("Deliver Video {$video->id}")
91-
->allowFailures()
92-
->onQueue('activitypub-out')
93-
->dispatch();
94-
95-
if (config('logging.dev_log')) {
96-
Log::info('Video creation dispatched', [
97-
'video_id' => $video->id,
98-
'inbox_count' => count($jobs),
99-
'total_recipients' => collect($jobs)->sum(fn ($job) => count($job->recipientProfileIds)),
100-
]);
101+
if ($actor->local && $video->visibility == 1) {
102+
app(\App\Services\RelayService::class)->deliverToRelays($actor, \App\Federation\ActivityBuilders\CreateActivityBuilder::buildForVideo($actor, $video));
101103
}
102104
}
103105

0 commit comments

Comments
 (0)