From 80f792038ce1907448304f84c6990cbe705d04bd Mon Sep 17 00:00:00 2001 From: Daniel Supernault Date: Sat, 19 Sep 2026 02:54:29 -0600 Subject: [PATCH] Fix federation comment threading --- .../Commands/FixBugs/FixOrphanedReplies.php | 247 +++++++ app/Console/Commands/README.md | 1 + .../RemoteReplyResolvePipeline.php | 154 +++++ .../StatusPipeline/RemoteStatusDelete.php | 3 +- app/Jobs/StatusPipeline/StatusDelete.php | 3 +- app/Models/Status.php | 23 + app/Services/Status/ReplyCleanupService.php | 46 ++ .../ActivityPub/StatusTransformer.php | 10 +- .../ActivityPub/Verb/CreateNote.php | 2 +- app/Transformer/ActivityPub/Verb/Note.php | 2 +- app/Transformer/ActivityPub/Verb/Question.php | 2 +- .../ActivityPub/Verb/UpdateNote.php | 2 +- app/Util/ActivityPub/Helpers.php | 368 ++++++++-- app/Util/ActivityPub/Inbox/HandlesCreates.php | 135 +++- .../Console/FixOrphanedRepliesTest.php | 178 +++++ .../Federation/RemoteReplyThreadingTest.php | 627 ++++++++++++++++++ .../Feature/StatusDeleteReplyCleanupTest.php | 122 ++++ 17 files changed, 1845 insertions(+), 80 deletions(-) create mode 100644 app/Console/Commands/FixBugs/FixOrphanedReplies.php create mode 100644 app/Jobs/StatusPipeline/RemoteReplyResolvePipeline.php create mode 100644 app/Services/Status/ReplyCleanupService.php create mode 100644 tests/Feature/Console/FixOrphanedRepliesTest.php create mode 100644 tests/Feature/Federation/RemoteReplyThreadingTest.php create mode 100644 tests/Feature/StatusDeleteReplyCleanupTest.php diff --git a/app/Console/Commands/FixBugs/FixOrphanedReplies.php b/app/Console/Commands/FixBugs/FixOrphanedReplies.php new file mode 100644 index 000000000..f19bcb245 --- /dev/null +++ b/app/Console/Commands/FixBugs/FixOrphanedReplies.php @@ -0,0 +1,247 @@ + 0, + 'top_level' => 0, + 'unreachable' => 0, + 'orphans' => 0, + 'parent_known' => 0, + 'relinked' => 0, + 'pruned' => 0, + 'unrecoverable' => 0, + ]; + + public function handle(): int + { + $days = max(1, (int) $this->option('days')); + $limit = max(0, (int) $this->option('limit')); + $delay = max(0, (int) $this->option('delay')); + $fix = (bool) $this->option('fix'); + $prune = (bool) $this->option('prune'); + + if ($prune && ! $fix) { + $this->error('--prune only works together with --fix.'); + + return self::FAILURE; + } + + if ($prune && $this->input->isInteractive() && ! $this->confirm( + 'Orphans whose parent cannot be resolved right now will be deleted, including ones whose parent server is only temporarily down. Continue?' + )) { + return self::SUCCESS; + } + + $types = $this->option('media') + ? array_merge(['text'], self::MEDIA_TYPES) + : ['text']; + + // Status ids are snowflakes, so a date maps to an id and the scan is + // a bounded range on an index instead of a walk over created_at. + $minId = SnowflakeService::byDate(now()->subDays($days)); + + $this->info(sprintf( + '%s remote %s statuses without a parent from the last %d day(s)...', + $fix ? 'Repairing' : 'Checking (read-only)', + implode(', ', $types), + $days + )); + + Status::query() + ->where('id', '>=', $minId) + ->whereNotNull('uri') + ->whereNull('in_reply_to_id') + ->whereNull('reblog_of_id') + ->whereIn('type', $types) + ->whereIn('scope', ['public', 'unlisted', 'private']) + ->chunkById(200, function ($statuses) use ($limit, $delay, $fix, $prune) { + foreach ($statuses as $status) { + if ($limit && $this->stats['checked'] >= $limit) { + return false; + } + + $this->check($status, $fix, $prune); + + if ($delay) { + usleep($delay * 1000); + } + } + + return true; + }); + + $this->report($fix); + + return self::SUCCESS; + } + + private function check(Status $status, bool $fix, bool $prune): void + { + $this->stats['checked']++; + + $profile = $status->profile; + $source = $status->object_url ?: $status->uri; + + if (! $profile || $profile->domain === null || ! $source) { + $this->stats['unreachable']++; + + return; + } + + $object = Helpers::fetchFromUrl($source); + + if (is_array($object) && isset($object['object']) && is_array($object['object'])) { + $object = $object['object']; + } + + // Gone, private, or not the object we stored. Nothing to learn. + if ( + ! is_array($object) || + ($status->object_url && ($object['id'] ?? null) !== $status->object_url) + ) { + $this->stats['unreachable']++; + + return; + } + + $inReplyTo = $object['inReplyTo'] ?? null; + + if ($inReplyTo === null || $inReplyTo === '' || $inReplyTo === []) { + $this->stats['top_level']++; + + return; + } + + $this->stats['orphans']++; + + if (! $fix) { + $url = Helpers::pluckval($inReplyTo); + $known = is_string($url) && Helpers::findExistingStatus($url); + + if ($known) { + $this->stats['parent_known']++; + } + + $this->line(sprintf( + ' orphan %s (%s) parent %s', + $status->id, + $status->type, + $known ? 'is stored' : 'is not stored' + ), null, 'v'); + + return; + } + + $resolution = Helpers::resolveReplyParent($object, $profile); + $parent = $resolution['status']; + + if ( + $resolution['state'] === Helpers::REPLY_PARENT_RESOLVED && + $parent && + (string) $parent->id !== (string) $status->id + ) { + $this->relink($status, $parent); + + return; + } + + if ($prune) { + RemoteStatusDelete::dispatch($status)->onQueue('delete'); + $this->stats['pruned']++; + + return; + } + + $this->stats['unrecoverable']++; + } + + /** + * Attach the orphan to its parent and take it out of the places a + * top-level post lives. No notification: the reply may be weeks old. + */ + private function relink(Status $status, Status $parent): void + { + $status->in_reply_to_id = $parent->id; + $status->in_reply_to_profile_id = $parent->profile_id; + $status->save(); + + $parent->reply_count = ($parent->reply_count ?? 0) + 1; + $parent->save(); + + StatusService::del($status->id); + StatusService::del($parent->id); + Cache::forget('status:replies:all:'.$parent->id); + + if (in_array($status->type, self::MEDIA_TYPES, true)) { + NetworkTimelineService::del($status->id); + FeedRemoveRemotePipeline::dispatch($status->id, $status->profile_id)->onQueue('feed'); + } + + $this->stats['relinked']++; + + $this->line(" relinked {$status->id} to {$parent->id}", null, 'v'); + } + + private function report(bool $fix): void + { + $rows = [ + ['Candidates checked', $this->stats['checked']], + ['Really top-level', $this->stats['top_level']], + ['Source unreachable, skipped', $this->stats['unreachable']], + ['Orphaned replies found', $this->stats['orphans']], + ]; + + if ($fix) { + $rows[] = ['Relinked', $this->stats['relinked']]; + $rows[] = ['Deleted (--prune)', $this->stats['pruned']]; + $rows[] = ['Left alone, parent unavailable', $this->stats['unrecoverable']]; + } else { + $rows[] = [' parent already stored', $this->stats['parent_known']]; + $rows[] = [' parent not stored', $this->stats['orphans'] - $this->stats['parent_known']]; + } + + $this->newLine(); + $this->table(['', 'Count'], $rows); + + if (! $fix && $this->stats['orphans']) { + $this->line('Run again with --fix to relink them. Add --prune to delete the ones that cannot be relinked.'); + } + } +} diff --git a/app/Console/Commands/README.md b/app/Console/Commands/README.md index 3c1899909..bf5612db5 100644 --- a/app/Console/Commands/README.md +++ b/app/Console/Commands/README.md @@ -69,6 +69,7 @@ full argument and option list. | `fix:profile:duplicates` | Fix duplicate profiles. | | `fix:hashtags` | Fix hashtag records. | | `fix:likes` | Recalculate like counts. | +| `fix:orphaned-replies` | Find remote replies stored without a parent and relink them (`--fix`), optionally deleting unrecoverable ones (`--prune`). Read-only by default. | | `media:fix-nonlocal-driver` | Repair filesystem records when `FILESYSTEM_DRIVER` is not set to local. | | `app:fix-missing-user-profile` | Interactively create a missing profile for an affected user. | | `fix:usernames` | Fix invalid usernames. | diff --git a/app/Jobs/StatusPipeline/RemoteReplyResolvePipeline.php b/app/Jobs/StatusPipeline/RemoteReplyResolvePipeline.php new file mode 100644 index 000000000..943d236a8 --- /dev/null +++ b/app/Jobs/StatusPipeline/RemoteReplyResolvePipeline.php @@ -0,0 +1,154 @@ +id, 0) + ->delay(now()->addSeconds(self::BACKOFF[0])) + ->onQueue('low'); + + return true; + } + + public static function pendingKey(string $id): string + { + return 'pf:ap:reply-resolve:pending:'.hash('sha256', $id); + } + + public function handle(): void + { + $id = Helpers::pluckval($this->object['id'] ?? null); + + if (! is_string($id) || ! Helpers::validateUrl($id)) { + return; + } + + $profile = Profile::find($this->profileId); + + // Author deleted, suspended or somehow local: nothing to store. + if (! $profile || $profile->domain === null || $profile->status !== null) { + $this->finish($id); + + return; + } + + // Stored in the meantime by another path (Announce, Like, search). + if (Helpers::findExistingStatus($id)) { + $this->finish($id); + + return; + } + + $resolution = Helpers::resolveReplyParent($this->object, $profile); + + if ($resolution['state'] === Helpers::REPLY_PARENT_UNRESOLVED) { + $this->retryOrDrop($id); + + return; + } + + if (Helpers::replyParentAllowsStore($resolution)) { + Helpers::storeStatus($id, $profile, $this->object); + } + + $this->finish($id); + } + + private function retryOrDrop(string $id): void + { + $next = $this->attempt + 1; + + if (! isset(self::BACKOFF[$next])) { + Log::info('RemoteReplyResolvePipeline: parent never resolved, dropping reply', [ + 'id' => $id, + 'inReplyTo' => $this->object['inReplyTo'] ?? null, + 'attempts' => $next, + ]); + + $this->finish($id); + + return; + } + + self::dispatch($this->object, $this->profileId, $next) + ->delay(now()->addSeconds(self::BACKOFF[$next])) + ->onQueue('low'); + } + + private function finish(string $id): void + { + Cache::forget(self::pendingKey($id)); + } +} diff --git a/app/Jobs/StatusPipeline/RemoteStatusDelete.php b/app/Jobs/StatusPipeline/RemoteStatusDelete.php index 35c5f6f26..6e0c40f35 100644 --- a/app/Jobs/StatusPipeline/RemoteStatusDelete.php +++ b/app/Jobs/StatusPipeline/RemoteStatusDelete.php @@ -21,6 +21,7 @@ use App\Services\Account\AccountStatService; use App\Services\AccountService; use App\Services\CollectionService; use App\Services\NotificationService; +use App\Services\Status\ReplyCleanupService; use App\Services\StatusService; use Illuminate\Bus\Queueable; use Illuminate\Contracts\Queue\ShouldBeUniqueUntilProcessing; @@ -191,7 +192,7 @@ class RemoteStatusDelete implements ShouldBeUniqueUntilProcessing, ShouldQueue // decrements hashtags.cached_count (a query-builder delete bypasses it). StatusHashtag::whereStatusId($status->id)->get()->each->delete(); StatusView::whereStatusId($status->id)->delete(); - Status::whereInReplyToId($status->id)->update(['in_reply_to_id' => null]); + ReplyCleanupService::releaseRepliesOf($status); StatusService::del($status->id, true); AccountService::del($status->profile_id); diff --git a/app/Jobs/StatusPipeline/StatusDelete.php b/app/Jobs/StatusPipeline/StatusDelete.php index 183f237be..eefd6cbdf 100644 --- a/app/Jobs/StatusPipeline/StatusDelete.php +++ b/app/Jobs/StatusPipeline/StatusDelete.php @@ -23,6 +23,7 @@ use App\Services\ActivityPubDeliveryService; use App\Services\CollectionService; use App\Services\FractalService; use App\Services\NotificationService; +use App\Services\Status\ReplyCleanupService; use App\Services\StatusService; use App\Transformer\ActivityPub\Verb\DeleteNote; use Illuminate\Bus\Queueable; @@ -188,7 +189,7 @@ class StatusDelete implements ShouldQueue // decrements hashtags.cached_count (a query-builder delete bypasses it). StatusHashtag::whereStatusId($status->id)->get()->each->delete(); StatusView::whereStatusId($status->id)->delete(); - Status::whereInReplyToId($status->id)->update(['in_reply_to_id' => null]); + ReplyCleanupService::releaseRepliesOf($status); AccountInterstitial::where('item_type', Status::class) ->where('item_id', $status->id) diff --git a/app/Models/Status.php b/app/Models/Status.php index 3947c1c19..e8ba41c6b 100644 --- a/app/Models/Status.php +++ b/app/Models/Status.php @@ -285,6 +285,29 @@ class Status extends Model return false; } + /** + * ActivityPub id of the status this one replies to, for use as inReplyTo. + * + * For a remote parent this is object_url (the object's id), not uri. + * uri holds the object's `url` property, which for most software is the + * HTML permalink. Servers that thread by id cannot match a permalink, so + * sending it detaches our reply from the thread on their side. + */ + public function inReplyToUri(): ?string + { + if (! $this->in_reply_to_id) { + return null; + } + + $parent = self::find($this->in_reply_to_id); + + if (! $parent) { + return null; + } + + return $parent->object_url ?: $parent->url(); + } + public function conversation() { return $this->hasOne(Conversation::class); diff --git a/app/Services/Status/ReplyCleanupService.php b/app/Services/Status/ReplyCleanupService.php new file mode 100644 index 000000000..7224e460c --- /dev/null +++ b/app/Services/Status/ReplyCleanupService.php @@ -0,0 +1,46 @@ +id) + ->whereNotNull('uri') + ->chunkById(200, function ($replies) { + foreach ($replies as $reply) { + RemoteStatusDelete::dispatch($reply)->onQueue('delete'); + } + }); + + // Local replies are a local user's content and are not ours to + // remove because someone else deleted the post above them. They are + // detached, as before. Feeds and profile grids only list media + // types, so a text comment does not surface as a post. A reply that + // carries media (possible through the Mastodon API) does end up on + // its author's own profile, which is acceptable for their own media. + Status::whereInReplyToId($parent->id) + ->whereNull('uri') + ->update(['in_reply_to_id' => null]); + } +} diff --git a/app/Transformer/ActivityPub/StatusTransformer.php b/app/Transformer/ActivityPub/StatusTransformer.php index a471bf3be..ccd67ea36 100644 --- a/app/Transformer/ActivityPub/StatusTransformer.php +++ b/app/Transformer/ActivityPub/StatusTransformer.php @@ -4,7 +4,6 @@ namespace App\Transformer\ActivityPub; use App\Models\Status; use App\Services\MediaService; -use App\Services\StatusService; use App\Util\Lexer\Autolink; use League\Fractal; @@ -14,14 +13,7 @@ class StatusTransformer extends Fractal\TransformerAbstract { $content = $status->caption ? nl2br(Autolink::create()->autolink($status->caption)) : ''; - $inReplyTo = null; - - if ($status->in_reply_to_id) { - $reply = StatusService::get($status->in_reply_to_id, true); - if ($reply && isset($reply['url'])) { - $inReplyTo = $reply['url']; - } - } + $inReplyTo = $status->inReplyToUri(); return [ '@context' => [ diff --git a/app/Transformer/ActivityPub/Verb/CreateNote.php b/app/Transformer/ActivityPub/Verb/CreateNote.php index c16338f64..22a595dcd 100644 --- a/app/Transformer/ActivityPub/Verb/CreateNote.php +++ b/app/Transformer/ActivityPub/Verb/CreateNote.php @@ -102,7 +102,7 @@ class CreateNote extends Fractal\TransformerAbstract 'type' => 'Note', 'summary' => $status->is_nsfw ? $status->cw_summary : null, 'content' => $content, - 'inReplyTo' => $status->in_reply_to_id ? $status->parent()->url() : null, + 'inReplyTo' => $status->inReplyToUri(), 'published' => $status->created_at->toAtomString(), 'url' => $status->url(), 'attributedTo' => $status->profile->permalink(), diff --git a/app/Transformer/ActivityPub/Verb/Note.php b/app/Transformer/ActivityPub/Verb/Note.php index 91d974508..eaceb698d 100644 --- a/app/Transformer/ActivityPub/Verb/Note.php +++ b/app/Transformer/ActivityPub/Verb/Note.php @@ -96,7 +96,7 @@ class Note extends Fractal\TransformerAbstract 'type' => 'Note', 'summary' => $status->is_nsfw ? $status->cw_summary : null, 'content' => $content, - 'inReplyTo' => $status->in_reply_to_id ? $status->parent()->url() : null, + 'inReplyTo' => $status->inReplyToUri(), 'published' => $status->created_at->toAtomString(), 'url' => $status->url(), 'attributedTo' => $status->profile->permalink(), diff --git a/app/Transformer/ActivityPub/Verb/Question.php b/app/Transformer/ActivityPub/Verb/Question.php index cc86bb7e2..4d2dbc0f4 100644 --- a/app/Transformer/ActivityPub/Verb/Question.php +++ b/app/Transformer/ActivityPub/Verb/Question.php @@ -73,7 +73,7 @@ class Question extends Fractal\TransformerAbstract 'type' => 'Question', 'summary' => null, 'content' => $content, - 'inReplyTo' => $status->in_reply_to_id ? $status->parent()->url() : null, + 'inReplyTo' => $status->inReplyToUri(), 'published' => $status->created_at->toAtomString(), 'url' => $status->url(), 'attributedTo' => $status->profile->permalink(), diff --git a/app/Transformer/ActivityPub/Verb/UpdateNote.php b/app/Transformer/ActivityPub/Verb/UpdateNote.php index 11b97316f..cf9cc378a 100644 --- a/app/Transformer/ActivityPub/Verb/UpdateNote.php +++ b/app/Transformer/ActivityPub/Verb/UpdateNote.php @@ -103,7 +103,7 @@ class UpdateNote extends Fractal\TransformerAbstract 'type' => 'Note', 'summary' => $status->is_nsfw ? $status->cw_summary : null, 'content' => $content, - 'inReplyTo' => $status->in_reply_to_id ? $status->parent()->url() : null, + 'inReplyTo' => $status->inReplyToUri(), 'published' => $status->created_at->toAtomString(), 'url' => $status->url(), 'attributedTo' => $status->profile->permalink(), diff --git a/app/Util/ActivityPub/Helpers.php b/app/Util/ActivityPub/Helpers.php index 1544cb4e7..55109bf42 100644 --- a/app/Util/ActivityPub/Helpers.php +++ b/app/Util/ActivityPub/Helpers.php @@ -15,6 +15,7 @@ use App\Models\Poll; use App\Models\Profile; use App\Models\Status; use App\Services\Account\AccountStatService; +use App\Services\AccountService; use App\Services\ActivityPubDeliveryService; use App\Services\ActivityPubFetchService; use App\Services\DomainService; @@ -43,6 +44,35 @@ class Helpers private const int FETCH_CACHE_TTL = 15; + /** + * Seconds a failed fetch is remembered. Kept short on purpose: a failed + * parent fetch used to be cached for the same 15 minutes as a success, + * which pinned every reply to that parent as unresolvable. Two minutes + * still absorbs a burst of activities pointing at the same dead URL. + * RemoteReplyResolvePipeline::BACKOFF must start above this value. + */ + public const int FETCH_NEGATIVE_TTL = 120; + + /** + * Outcomes of resolveReplyParent(). + * + * NONE the object is not a reply + * RESOLVED the parent exists locally (it may have just been fetched) + * UNRESOLVED the object is a reply but the parent could not be found or + * fetched right now. Never store the object in this state, + * it would become a top-level status. Retry later. + * REJECTED the object is a reply we must not accept (blocked author, + * blocked domain, comments disabled, malformed inReplyTo). + * Never store, never retry. + */ + public const string REPLY_PARENT_NONE = 'none'; + + public const string REPLY_PARENT_RESOLVED = 'resolved'; + + public const string REPLY_PARENT_UNRESOLVED = 'unresolved'; + + public const string REPLY_PARENT_REJECTED = 'rejected'; + private const int MAX_URL_LENGTH = 4096; private const int DNS_TTL_POSITIVE = 86400; @@ -529,22 +559,56 @@ class Helpers return; } - $hash = hash('sha256', $url); - $key = "helpers:url:fetcher:sha256-{$hash}"; - $ttl = now()->addMinutes(15); + if (is_array($url)) { + $url = $url[0] ?? null; + } - return Cache::remember($key, $ttl, function () use ($url) { - $res = ActivityPubFetchService::get($url); - if (! $res || empty($res)) { - return false; - } - $res = json_decode($res, true, 8); - if (json_last_error() === JSON_ERROR_NONE) { - return $res; - } + if (! is_string($url)) { + return; + } + + $key = self::fetchCacheKey($url); + $cached = Cache::get($key); + + if ($cached !== null) { + return $cached; + } + + $res = self::fetchAndDecode($url); + + // Successes and failures get different lifetimes, see FETCH_NEGATIVE_TTL. + Cache::put( + $key, + $res, + $res === false + ? self::FETCH_NEGATIVE_TTL + : now()->addMinutes(self::FETCH_CACHE_TTL) + ); + + return $res; + } + + public static function fetchCacheKey(string $url): string + { + return 'helpers:url:fetcher:sha256-'.hash('sha256', $url); + } + + private static function fetchAndDecode(string $url): array|false + { + $res = ActivityPubFetchService::get($url); + + if (! $res || empty($res)) { return false; - }); + } + + $res = json_decode($res, true, 8); + + if (json_last_error() !== JSON_ERROR_NONE || ! is_array($res)) { + return false; + } + + return $res; } public static function fetchProfileFromUrl($url) @@ -599,6 +663,15 @@ class Helpers return $status; } + // A local URL we cannot map to a row is a deleted, archived or + // unknown status. There is nothing to fetch, and fetching our own + // domain would store a remote copy of a local status. + $host = parse_url($url, PHP_URL_HOST); + + if (is_string($host) && self::isLocalDomain($host)) { + return null; + } + // Bound how far up an inReplyTo chain a single fetch may walk. // Checked after the DB lookup so a reply to an already-known // status still links even at the limit, but we never fetch past it. @@ -697,6 +770,10 @@ class Helpers { $host = parse_url($url, PHP_URL_HOST); + if (! is_string($host) || $host === '') { + return null; + } + if (self::isLocalDomain($host)) { $id = self::extractLocalStatusId($url); @@ -753,19 +830,22 @@ class Helpers $cw = self::getSensitive($object, $url); if (($object['type'] ?? null) === 'Question') { - $replyToId = self::getReplyToId( + $resolution = self::resolveReplyParent( $activity, $profile, - $replyTo, $depth ); + if (! self::replyParentAllowsStore($resolution)) { + return null; + } + return self::storePoll( $profile, $object, $url, $object['published'] ?? $res['published'], - $replyToId, + $resolution['status']?->id, $cw, $scope, $object['id'] ?? $url @@ -926,51 +1006,169 @@ class Helpers } /** - * Get reply-to status ID + * Resolve the parent of a reply. + * + * Resolves (and fetches if needed) the status referenced by + * object.inReplyTo, one hop deeper than the caller, and reports why when + * it cannot. "This is not a reply" and "this is a reply whose parent we + * could not get" are different answers: only the first may be stored + * without in_reply_to_id. * - * Resolves (and fetches if needed) the parent referenced by - * object.inReplyTo, one hop deeper than the caller. + * @return array{state: string, status: ?Status} */ - public static function getReplyToId( + public static function resolveReplyParent( array $activity, Profile $profile, - bool $replyTo = false, int $depth = 0 - ): ?int { + ): array { $object = self::statusObject($activity); - $inReplyTo = self::pluckval($object['inReplyTo'] ?? null); + $raw = $object['inReplyTo'] ?? null; - if (! is_string($inReplyTo) || $inReplyTo === '') { - return null; + if ($raw === null || $raw === '' || $raw === []) { + return self::replyParentResult(self::REPLY_PARENT_NONE); } - $reply = self::statusFirstOrFetch( + $inReplyTo = self::extractInReplyTo($raw); + + if ($inReplyTo === null) { + // Declared as a reply, but to nothing we can address. + return self::replyParentResult(self::REPLY_PARENT_REJECTED); + } + + $parent = self::statusFirstOrFetch( $inReplyTo, false, $depth + 1 ); - if (! $reply) { + if (! $parent) { + return self::replyParentResult(self::REPLY_PARENT_UNRESOLVED); + } + + if (self::parentRefusesReplyFrom($parent, $profile)) { + return self::replyParentResult(self::REPLY_PARENT_REJECTED); + } + + return self::replyParentResult(self::REPLY_PARENT_RESOLVED, $parent); + } + + /** + * @return array{state: string, status: ?Status} + */ + private static function replyParentResult(string $state, ?Status $status = null): array + { + return ['state' => $state, 'status' => $status]; + } + + /** + * Whether an object with this resolution may be written to the database. + * + * @param array{state: string, status: ?Status} $resolution + */ + public static function replyParentAllowsStore(array $resolution): bool + { + return in_array($resolution['state'], [ + self::REPLY_PARENT_NONE, + self::REPLY_PARENT_RESOLVED, + ], true); + } + + /** + * Pull a single URL out of an inReplyTo value. JSON-LD allows a string, + * a list, an embedded object, or a Link. + */ + private static function extractInReplyTo(mixed $value): ?string + { + if (is_string($value)) { + $value = trim($value); + + return $value !== '' ? $value : null; + } + + if (! is_array($value) || $value === []) { return null; } - $blocks = UserFilterService::blocks($reply->profile_id); + if (array_is_list($value)) { + return self::extractInReplyTo($value[0]); + } + + foreach (['id', 'href'] as $key) { + if (isset($value[$key]) && is_string($value[$key]) && trim($value[$key]) !== '') { + return trim($value[$key]); + } + } + + return null; + } + + /** + * Whether the parent's author has opted out of replies from this profile. + * + * Ids are compared as strings: UserFilterService returns them from Redis + * as strings while model keys are integers, so a strict in_array() on the + * raw values never matches. + */ + private static function parentRefusesReplyFrom(Status $parent, Profile $profile): bool + { + if ((string) $parent->profile_id === (string) $profile->id) { + return false; + } + + $blocks = array_map( + 'strval', + (array) UserFilterService::blocks((int) $parent->profile_id) + ); + + if (in_array((string) $profile->id, $blocks, true)) { + return true; + } + + // Domain blocks and the comments toggle only exist for local authors. + // For a remote parent the origin server is the authority. + if (! empty($parent->uri)) { + return false; + } + + if ($parent->comments_disabled) { + return true; + } - return in_array($profile->id, $blocks, true) - ? null - : $reply->id; + return $profile->domain !== null && + AccountService::blocksDomain($parent->profile_id, $profile->domain) === true; + } + + /** + * Get reply-to status ID + * + * Kept for callers that only need the id. It cannot tell "not a reply" + * from "parent unavailable", so nothing that stores a status should use + * it. Use resolveReplyParent(). + */ + public static function getReplyToId( + array $activity, + Profile $profile, + bool $replyTo = false, + int $depth = 0 + ): ?int { + return self::resolveReplyParent($activity, $profile, $depth)['status']?->id; } /** * Store a new regular status + * + * Returns null, and writes nothing, when the object is a reply whose + * parent is unresolved or refuses it. A reply is never stored without + * in_reply_to_id. Inbox deliveries that want a retry should check + * resolveReplyParent() first, see HandlesCreates::handleNoteReply(). */ public static function storeStatus( string $url, Profile $profile, array $activity, int $depth = 0 - ): Status { + ): ?Status { $object = self::statusObject($activity); $id = self::getStatusId($object, $url); @@ -996,13 +1194,19 @@ class Helpers ])); } - $replyTo = self::getReplyToId( + $resolution = self::resolveReplyParent( ['object' => $object], $profile, - false, $depth ); + if (! self::replyParentAllowsStore($resolution)) { + return null; + } + + $parent = $resolution['status']; + $replyTo = $parent?->id; + $published = self::pluckval( $object['published'] ?? $activity['published'] @@ -1034,9 +1238,19 @@ class Helpers $replyTo, $cw, $scope, - $commentsDisabled + $commentsDisabled, + $parent?->profile_id ); + if (! $status) { + return null; + } + + // False when another worker stored this status first, or when this + // is a refresh of a known status. Counters, notifications and feed + // fan-out must only run once per status. + $isNew = $status->wasRecentlyCreated; + if ($replyTo === null) { self::importNoteAttachment($object, $status); } else { @@ -1047,7 +1261,9 @@ class Helpers self::importNoteAttachment($object, $status); } - StatusReplyPipeline::dispatch($status); + if ($isNew) { + StatusReplyPipeline::dispatch($status); + } } if ( @@ -1058,11 +1274,13 @@ class Helpers StatusTagsPipeline::dispatch($object, $status); } - self::handleStatusPostProcessing( - $status, - $profile->id, - $url - ); + if ($isNew) { + self::handleStatusPostProcessing( + $status, + $profile->id, + $url + ); + } return $status; } @@ -1116,6 +1334,9 @@ class Helpers /** * Create or update status record + * + * Returns null when the status exists but was deleted locally, or when + * its uri or object_url is already held by another profile. */ public static function createOrUpdateStatus( string $url, @@ -1126,8 +1347,9 @@ class Helpers ?int $reply_to, bool $cw, string $scope, - bool $commentsDisabled - ): Status { + bool $commentsDisabled, + ?int $replyToProfileId = null + ): ?Status { $caption = isset($activity['content']) ? app(SanitizeService::class)->html($activity['content']) : ''; @@ -1135,24 +1357,48 @@ class Helpers app(SanitizeService::class)->html($activity['summary']) : null; - return Status::updateOrCreate( - ['uri' => $url], - [ - 'profile_id' => $profile->id, - 'url' => $url, - 'object_url' => $id, - 'caption' => strip_tags($caption), - 'rendered' => $caption, - 'created_at' => Carbon::parse($ts)->tz('UTC'), - 'in_reply_to_id' => $reply_to, - 'local' => false, - 'is_nsfw' => $cw, - 'scope' => $scope, - 'visibility' => $scope, - 'cw_summary' => $cwSummary ? strip_tags($cwSummary) : null, - 'comments_disabled' => $commentsDisabled, - ] - ); + $attributes = [ + 'profile_id' => $profile->id, + 'url' => $url, + 'object_url' => $id, + 'caption' => strip_tags($caption), + 'rendered' => $caption, + 'created_at' => Carbon::parse($ts)->tz('UTC'), + 'in_reply_to_id' => $reply_to, + 'in_reply_to_profile_id' => $reply_to ? $replyToProfileId : null, + 'local' => false, + 'is_nsfw' => $cw, + 'scope' => $scope, + 'visibility' => $scope, + 'cw_summary' => $cwSummary ? strip_tags($cwSummary) : null, + 'comments_disabled' => $commentsDisabled, + ]; + + try { + return Status::updateOrCreate(['uri' => $url], $attributes); + } catch (UniqueConstraintViolationException) { + // uri and object_url are both unique. We land here when another + // worker inserted the same status between updateOrCreate's + // select and insert (two replies to one unknown parent arriving + // together), or when the row exists but is soft deleted. + // Returning the winner keeps the rest of the reply chain alive + // instead of failing the whole job. A soft deleted row means the + // status was removed and must not come back. + $existing = Status::withTrashed() + ->where(function ($query) use ($url, $id) { + $query->where('uri', $url) + ->orWhere('object_url', $id); + }) + ->first(); + + // Only hand back a row by the same author. A row that holds this + // uri or object_url under another profile is not this status. + return $existing && + ! $existing->trashed() && + (string) $existing->profile_id === (string) $profile->id + ? $existing + : null; + } } /** diff --git a/app/Util/ActivityPub/Inbox/HandlesCreates.php b/app/Util/ActivityPub/Inbox/HandlesCreates.php index d77f9c58c..c797ebddb 100644 --- a/app/Util/ActivityPub/Inbox/HandlesCreates.php +++ b/app/Util/ActivityPub/Inbox/HandlesCreates.php @@ -3,6 +3,7 @@ namespace App\Util\ActivityPub\Inbox; use App\Jobs\PushNotificationPipeline\MentionPushNotifyPipeline; +use App\Jobs\StatusPipeline\RemoteReplyResolvePipeline; use App\Models\Conversation; use App\Models\DirectMessage; use App\Models\Media; @@ -48,6 +49,20 @@ trait HandlesCreates return; } + // A poll vote is addressed to the poll author alone, which is also + // the shape of a direct message, so it has to be recognised first. + // Votes are tallies: never a DM, never a comment. + if ($activity['type'] == 'Note' && $this->isPollVoteObject($activity)) { + if ( + is_string($activity['inReplyTo']) && + Helpers::validateLocalUrl($activity['inReplyTo']) + ) { + $this->handlePollVote(); + } + + return; + } + if ($this->isDirectMessage($to, $cc)) { $this->handleDirectMessage(); @@ -64,16 +79,113 @@ trait HandlesCreates } } + /** + * Handle a delivered reply. + * + * The object is stored from the signed delivery, not re-fetched. Fetching + * loses every reply our instance actor cannot read (followers-only + * replies to local posts), and the old fetch used object.url, which is an + * HTML permalink on most software and an array on some. + * + * A reply is only ever stored with its parent attached. If the parent + * cannot be resolved right now the delivery is parked in + * RemoteReplyResolvePipeline and retried, never written as a top-level + * status. + */ public function handleNoteReply(): void { - $activity = $this->payload['object']; - $actor = $this->validateAndFetchActor($this->payload['actor']); + $object = $this->payload['object']; + + $actorUrl = Helpers::pluckval($this->payload['actor'] ?? null); + if (! is_string($actorUrl)) { + return; + } + + $actor = $this->validateAndFetchActor($actorUrl); if (! $actor || $actor->domain == null) { return; } - $url = $activity['url'] ?? $activity['id']; - Helpers::statusFirstOrFetch($url, true); + $id = Helpers::pluckval($object['id'] ?? null); + if (! is_string($id) || ! Helpers::validateUrl($id)) { + return; + } + + if (Helpers::findExistingStatus($id)) { + return; + } + + if (! $this->deliveredObjectIsTrusted($object, $actor, $id)) { + // Not provably authored by the sender. Ask the origin, by id. + Helpers::statusFirstOrFetch($id, true); + + return; + } + + $published = Helpers::pluckval($object['published'] ?? null); + if (! is_string($published) || ! Helpers::validateTimestamp($published)) { + return; + } + + $resolution = Helpers::resolveReplyParent($object, $actor); + + if ($resolution['state'] === Helpers::REPLY_PARENT_UNRESOLVED) { + RemoteReplyResolvePipeline::park($object, $actor); + + return; + } + + if (! Helpers::replyParentAllowsStore($resolution)) { + return; + } + + Helpers::storeStatus($id, $actor, $object); + } + + /** + * A delivered object can be stored as-is only when the sender is its + * author: attributedTo is exactly the activity actor, the object id lives + * on the actor's host, and, when the signing key maps to a known profile, + * that profile is the actor. InboxValidator only binds these by host. + */ + protected function deliveredObjectIsTrusted(array $object, Profile $actor, string $id): bool + { + $attributedTo = $object['attributedTo'] ?? null; + + if (! is_string($attributedTo) && ! is_array($attributedTo)) { + return false; + } + + $author = Helpers::extractAttributedTo($attributedTo); + + if (! is_string($author) || $author !== $actor->remote_url) { + return false; + } + + $idHost = parse_url($id, PHP_URL_HOST); + $actorHost = parse_url((string) $actor->remote_url, PHP_URL_HOST); + + if ( + ! is_string($idHost) || + ! is_string($actorHost) || + strcasecmp($idHost, $actorHost) !== 0 + ) { + return false; + } + + $signer = $this->signingProfile(); + + return ! $signer || (string) $signer->id === (string) $actor->id; + } + + /** + * A poll vote: a Note that names an option and says nothing else. + */ + protected function isPollVoteObject(array $object): bool + { + return isset($object['inReplyTo'], $object['name']) && + ! isset($object['content']) && + ! isset($object['attachment']); } public function handlePollCreate(): void @@ -128,6 +240,21 @@ trait HandlesCreates } } + $id = Helpers::pluckval($activity['id'] ?? null); + + if (! is_string($id) || ! Helpers::validateUrl($id)) { + return; + } + + // Same rule as replies: only store the delivered object when the + // sender is provably its author. Otherwise an actor could claim + // another server's object id and squat its unique object_url. + if (! $this->deliveredObjectIsTrusted($activity, $actor, $id)) { + Helpers::statusFirstOrFetch($id); + + return; + } + Helpers::storeStatus($url, $actor, $activity); } diff --git a/tests/Feature/Console/FixOrphanedRepliesTest.php b/tests/Feature/Console/FixOrphanedRepliesTest.php new file mode 100644 index 000000000..9a6460193 --- /dev/null +++ b/tests/Feature/Console/FixOrphanedRepliesTest.php @@ -0,0 +1,178 @@ +remote()->create([ + 'domain' => 'remote.example', + 'username' => '@bob@remote.example', + 'remote_url' => 'https://remote.example/users/bob', + 'last_fetched_at' => now(), + ]); +} + +/** + * A remote status stored as top-level, plus what its origin says it is. + */ +function orphanStatus(Profile $author, string $path, ?string $inReplyTo, string $type = 'text'): Status +{ + $id = "https://remote.example/users/bob/statuses/{$path}"; + + $status = Status::factory()->create([ + 'profile_id' => $author->id, + 'type' => $type, + 'uri' => "https://remote.example/@bob/{$path}", + 'url' => "https://remote.example/@bob/{$path}", + 'object_url' => $id, + 'local' => false, + ]); + + Cache::put(Helpers::fetchCacheKey($id), [ + '@context' => 'https://www.w3.org/ns/activitystreams', + 'id' => $id, + 'type' => 'Note', + 'attributedTo' => $author->remote_url, + 'content' => '

hi

', + 'published' => now()->subHour()->toAtomString(), + 'inReplyTo' => $inReplyTo, + 'to' => ['https://www.w3.org/ns/activitystreams#Public'], + ], 600); + + return $status; +} + +it('reports orphans without changing anything by default', function () { + $user = User::factory()->create(); + $user->refresh(); + $parent = Status::factory()->photo()->create(['profile_id' => $user->profile_id]); + $bob = orphanRemoteProfile(); + orphanSeedHosts(); + + $orphan = orphanStatus($bob, '1', $parent->url()); + $missing = orphanStatus($bob, '2', 'https://other.example/users/alice/statuses/unknown'); + + $this->artisan('fix:orphaned-replies') + ->expectsOutputToContain('Orphaned replies found') + ->assertSuccessful(); + + expect($orphan->fresh()->in_reply_to_id)->toBeNull(); + expect($missing->fresh()->in_reply_to_id)->toBeNull(); + expect(Status::count())->toBe(3); + Http::assertNothingSent(); + Queue::assertNotPushed(RemoteStatusDelete::class); + Queue::assertNotPushed(FeedRemoveRemotePipeline::class); +}); + +it('relinks an orphan to its parent with --fix', function () { + $user = User::factory()->create(); + $user->refresh(); + $parent = Status::factory()->photo()->create(['profile_id' => $user->profile_id, 'reply_count' => 2]); + $bob = orphanRemoteProfile(); + orphanSeedHosts(); + + $orphan = orphanStatus($bob, '3', $parent->url()); + $topLevel = orphanStatus($bob, '4', null); + + $this->artisan('fix:orphaned-replies', ['--fix' => true])->assertSuccessful(); + + $orphan->refresh(); + + expect((int) $orphan->in_reply_to_id)->toBe((int) $parent->id); + expect((int) $orphan->in_reply_to_profile_id)->toBe((int) $user->profile_id); + expect($parent->fresh()->reply_count)->toBe(3); + expect($topLevel->fresh()->in_reply_to_id)->toBeNull(); +}); + +it('pulls a relinked media orphan out of home feeds', function () { + $user = User::factory()->create(); + $user->refresh(); + $parent = Status::factory()->photo()->create(['profile_id' => $user->profile_id]); + $bob = orphanRemoteProfile(); + orphanSeedHosts(); + + $orphan = orphanStatus($bob, '5', $parent->url(), 'photo'); + + // Text only by default: a photo status is not a candidate. + $this->artisan('fix:orphaned-replies', ['--fix' => true])->assertSuccessful(); + expect($orphan->fresh()->in_reply_to_id)->toBeNull(); + + $this->artisan('fix:orphaned-replies', ['--fix' => true, '--media' => true])->assertSuccessful(); + + expect((int) $orphan->fresh()->in_reply_to_id)->toBe((int) $parent->id); + Queue::assertPushed(FeedRemoveRemotePipeline::class, 1); +}); + +it('leaves unrecoverable orphans alone unless --prune is given', function () { + $bob = orphanRemoteProfile(); + orphanSeedHosts(); + + $gone = 'https://other.example/users/alice/statuses/deleted'; + Cache::put(Helpers::fetchCacheKey($gone), false, 600); + $orphan = orphanStatus($bob, '6', $gone); + + $this->artisan('fix:orphaned-replies', ['--fix' => true])->assertSuccessful(); + + expect($orphan->fresh())->not->toBeNull(); + Queue::assertNotPushed(RemoteStatusDelete::class); + + $this->artisan('fix:orphaned-replies', ['--fix' => true, '--prune' => true, '--no-interaction' => true]) + ->assertSuccessful(); + + Queue::assertPushedOn('delete', RemoteStatusDelete::class); +}); + +it('skips a status whose origin no longer serves it', function () { + $bob = orphanRemoteProfile(); + orphanSeedHosts(); + + $status = orphanStatus($bob, '7', 'https://other.example/users/alice/statuses/x'); + Cache::put(Helpers::fetchCacheKey($status->object_url), false, 600); + + $this->artisan('fix:orphaned-replies', ['--fix' => true, '--prune' => true, '--no-interaction' => true]) + ->assertSuccessful(); + + expect($status->fresh())->not->toBeNull(); + Queue::assertNotPushed(RemoteStatusDelete::class); +}); + +it('refuses --prune without --fix', function () { + $this->artisan('fix:orphaned-replies', ['--prune' => true])->assertFailed(); +}); diff --git a/tests/Feature/Federation/RemoteReplyThreadingTest.php b/tests/Feature/Federation/RemoteReplyThreadingTest.php new file mode 100644 index 000000000..6b7166e39 --- /dev/null +++ b/tests/Feature/Federation/RemoteReplyThreadingTest.php @@ -0,0 +1,627 @@ + false, + 'federation.activitypub.enabled' => true, + ]); +}); + +function replyLocalUser(): User +{ + $user = User::factory()->create(); + $user->refresh(); + + return $user; +} + +function replyRemoteProfile(string $domain = 'remote.example', string $username = 'bob'): Profile +{ + $actor = "https://{$domain}/users/{$username}"; + + return Profile::factory()->remote()->create([ + 'domain' => $domain, + 'username' => "@{$username}@{$domain}", + 'remote_url' => $actor, + 'key_id' => "{$actor}#main-key", + 'inbox_url' => "{$actor}/inbox", + 'sharedInbox' => "https://{$domain}/inbox", + 'last_fetched_at' => now(), + ]); +} + +/** + * Seed the DNS and banned-domain caches so URL validation passes without a + * network lookup. Call after factories, the lazy refresh can flush the cache. + */ +function replySeedHosts(array $hosts = ['remote.example', 'other.example']): void +{ + $hosts[] = config('pixelfed.domain.app'); + + foreach ($hosts as $host) { + Cache::put('helpers:url:public-ips:'.hash('xxh128', $host), ['203.0.113.40'], 3600); + } + + Cache::put('instances:banned:domains', [], 1209600); +} + +/** + * Make $url dereference to $object, or to a failed fetch when $object is + * false, by seeding the cache fetchFromUrl() reads. + */ +function replySeedFetch(string $url, array|false $object): void +{ + Cache::put(Helpers::fetchCacheKey($url), $object, 600); +} + +function replyNote(Profile $author, string $path, ?string $inReplyTo, array $overrides = []): array +{ + $id = $author->remote_url.'/statuses/'.$path; + + return array_merge([ + 'id' => $id, + 'type' => 'Note', + 'attributedTo' => $author->remote_url, + 'url' => "https://{$author->domain}/@user/{$path}", + 'content' => '

hello

', + 'published' => now()->subMinute()->toAtomString(), + 'inReplyTo' => $inReplyTo, + 'to' => ['https://www.w3.org/ns/activitystreams#Public'], + 'cc' => [$author->remote_url.'/followers'], + ], $overrides); +} + +function replyDeliver(Profile $actor, array $object, ?Profile $signer = null): void +{ + $payload = [ + '@context' => 'https://www.w3.org/ns/activitystreams', + 'id' => $object['id'].'/activity', + 'type' => 'Create', + 'actor' => $actor->remote_url, + 'object' => $object, + ]; + + $headers = [ + 'signature' => ['keyId="'.($signer ?? $actor)->key_id.'",algorithm="rsa-sha256",headers="(request-target) host date digest",signature="dGVzdA=="'], + 'date' => [now()->toRfc7231String()], + ]; + + (new Inbox($headers, null, $payload))->handle(); +} + +describe('storing', function () { + it('links a delivered reply to a local parent and records the parent author', function () { + $local = replyLocalUser(); + $parent = Status::factory()->photo()->create(['profile_id' => $local->profile_id]); + $bob = replyRemoteProfile(); + replySeedHosts(); + + replyDeliver($bob, replyNote($bob, '1', $parent->url())); + + $reply = Status::whereObjectUrl($bob->remote_url.'/statuses/1')->first(); + + expect($reply)->not->toBeNull(); + expect((int) $reply->in_reply_to_id)->toBe((int) $parent->id); + expect((int) $reply->in_reply_to_profile_id)->toBe((int) $local->profile_id); + + Queue::assertPushed(StatusReplyPipeline::class, 1); + Queue::assertNotPushed(RemoteReplyResolvePipeline::class); + }); + + it('stores the delivered object without fetching it back', function () { + $local = replyLocalUser(); + $parent = Status::factory()->photo()->create(['profile_id' => $local->profile_id]); + $bob = replyRemoteProfile(); + replySeedHosts(); + + // A followers-only reply: the origin would 404 our instance actor. + $note = replyNote($bob, '2', $parent->url(), [ + 'to' => [$bob->remote_url.'/followers'], + 'cc' => [$local->profile->permalink()], + ]); + replySeedFetch($note['id'], false); + replySeedFetch($note['url'], false); + + replyDeliver($bob, $note); + + $reply = Status::whereObjectUrl($note['id'])->first(); + + expect($reply)->not->toBeNull(); + expect($reply->scope)->toBe('private'); + expect((int) $reply->in_reply_to_id)->toBe((int) $parent->id); + Http::assertNothingSent(); + }); + + it('never stores a reply as a top-level status when the parent is unavailable', function () { + $bob = replyRemoteProfile(); + replySeedHosts(); + + $missing = 'https://other.example/users/alice/statuses/gone'; + replySeedFetch($missing, false); + + $note = replyNote($bob, '3', $missing, [ + 'attachment' => [[ + 'type' => 'Document', + 'mediaType' => 'image/jpeg', + 'url' => 'https://remote.example/media/1.jpg', + ]], + ]); + + expect(Helpers::storeStatus($note['id'], $bob, $note))->toBeNull(); + expect(Status::whereObjectUrl($note['id'])->exists())->toBeFalse(); + }); + + it('stores nothing from a chain that runs past the depth limit', function () { + $alice = replyRemoteProfile('other.example', 'alice'); + $bob = replyRemoteProfile(); + replySeedHosts(); + + // c1 <- c2 <- ... <- c8, none of them known locally, c1 replies to + // something that never resolves. The old code stored the ancestor at + // the limit as a root with in_reply_to_id null. + $previous = 'https://other.example/users/alice/statuses/c0'; + replySeedFetch($previous, false); + + foreach (range(1, 8) as $i) { + $note = replyNote($alice, "c{$i}", $previous); + replySeedFetch($note['id'], ['@context' => 'https://www.w3.org/ns/activitystreams'] + $note); + $previous = $note['id']; + } + + replyDeliver($bob, replyNote($bob, '4', $previous)); + + expect(Status::count())->toBe(0); + Queue::assertPushed(RemoteReplyResolvePipeline::class, 1); + }); + + it('fetches and links an unknown remote parent', function () { + $alice = replyRemoteProfile('other.example', 'alice'); + $bob = replyRemoteProfile(); + replySeedHosts(); + + $root = replyNote($alice, 'root', null); + replySeedFetch($root['id'], ['@context' => 'https://www.w3.org/ns/activitystreams'] + $root); + + replyDeliver($bob, replyNote($bob, '5', $root['id'])); + + $parent = Status::whereObjectUrl($root['id'])->first(); + $reply = Status::whereObjectUrl($bob->remote_url.'/statuses/5')->first(); + + expect($parent)->not->toBeNull(); + expect($parent->in_reply_to_id)->toBeNull(); + expect((int) $reply->in_reply_to_id)->toBe((int) $parent->id); + expect((int) $reply->in_reply_to_profile_id)->toBe((int) $alice->id); + }); + + it('does not count or notify twice when the same reply is stored again', function () { + $local = replyLocalUser(); + $parent = Status::factory()->photo()->create(['profile_id' => $local->profile_id]); + $bob = replyRemoteProfile(); + replySeedHosts(); + + $note = replyNote($bob, '6', $parent->url()); + + $first = Helpers::storeStatus($note['id'], $bob, $note); + $second = Helpers::storeStatus($note['id'], $bob, $note); + + expect((int) $second->id)->toBe((int) $first->id); + Queue::assertPushed(StatusReplyPipeline::class, 1); + }); + + it('returns the existing row when it loses an insert race', function () { + $bob = replyRemoteProfile(); + replySeedHosts(); + + $note = replyNote($bob, '7', null); + + // Same object_url under a different uri: updateOrCreate keys on uri, + // misses, inserts, and hits the unique index on object_url. This is + // what a second worker sees when it loses the race. + $winner = Status::factory()->create([ + 'profile_id' => $bob->id, + 'uri' => $note['id'], + 'url' => $note['id'], + 'object_url' => $note['id'], + 'local' => false, + ]); + + $status = Helpers::createOrUpdateStatus( + $note['url'], $bob, $note['id'], $note, $note['published'], + null, false, 'public', false + ); + + expect((int) $status->id)->toBe((int) $winner->id); + expect(Status::whereObjectUrl($note['id'])->count())->toBe(1); + }); +}); + +describe('refusals', function () { + it('drops a reply from a profile the parent author blocks', function () { + $local = replyLocalUser(); + $parent = Status::factory()->photo()->create(['profile_id' => $local->profile_id]); + $bob = replyRemoteProfile(); + replySeedHosts(); + + // Redis hands ids back as strings, model keys are integers. + Redis::shouldReceive('zrevrange')->andReturn([(string) $bob->id]); + + replyDeliver($bob, replyNote($bob, '8', $parent->url())); + + expect(Status::whereObjectUrl($bob->remote_url.'/statuses/8')->exists())->toBeFalse(); + Queue::assertNotPushed(RemoteReplyResolvePipeline::class); + }); + + it('drops a reply from a domain the parent author blocks', function () { + $local = replyLocalUser(); + $parent = Status::factory()->photo()->create(['profile_id' => $local->profile_id]); + $bob = replyRemoteProfile(); + replySeedHosts(); + + UserDomainBlock::create(['profile_id' => $local->profile_id, 'domain' => 'remote.example']); + + replyDeliver($bob, replyNote($bob, '9', $parent->url())); + + expect(Status::whereObjectUrl($bob->remote_url.'/statuses/9')->exists())->toBeFalse(); + Queue::assertNotPushed(RemoteReplyResolvePipeline::class); + }); + + it('drops a reply to a local status with comments disabled', function () { + $local = replyLocalUser(); + $parent = Status::factory()->photo()->create([ + 'profile_id' => $local->profile_id, + 'comments_disabled' => true, + ]); + $bob = replyRemoteProfile(); + replySeedHosts(); + + replyDeliver($bob, replyNote($bob, '10', $parent->url())); + + expect(Status::whereObjectUrl($bob->remote_url.'/statuses/10')->exists())->toBeFalse(); + }); + + it('does not self-fetch a local parent that no longer exists', function () { + $bob = replyRemoteProfile(); + replySeedHosts(); + + $gone = config('app.url').'/p/someone/999999999'; + + $result = Helpers::resolveReplyParent(replyNote($bob, '11', $gone), $bob); + + expect($result['state'])->toBe(Helpers::REPLY_PARENT_UNRESOLVED); + Http::assertNothingSent(); + }); + + it('rejects a reply whose inReplyTo cannot be read as a URL', function () { + $bob = replyRemoteProfile(); + replySeedHosts(); + + $result = Helpers::resolveReplyParent( + replyNote($bob, '12', null, ['inReplyTo' => ['type' => 'Note']]), + $bob + ); + + expect($result['state'])->toBe(Helpers::REPLY_PARENT_REJECTED); + }); + + it('reads inReplyTo given as a list or an embedded object', function () { + $local = replyLocalUser(); + $parent = Status::factory()->photo()->create(['profile_id' => $local->profile_id]); + $bob = replyRemoteProfile(); + replySeedHosts(); + + foreach ([[$parent->url()], ['type' => 'Note', 'id' => $parent->url()]] as $shape) { + $result = Helpers::resolveReplyParent( + replyNote($bob, '13', null, ['inReplyTo' => $shape]), + $bob + ); + + expect($result['state'])->toBe(Helpers::REPLY_PARENT_RESOLVED); + expect((int) $result['status']->id)->toBe((int) $parent->id); + } + }); +}); + +describe('delivery trust', function () { + it('does not store a delivered object attributed to someone other than the sender', function () { + $local = replyLocalUser(); + $parent = Status::factory()->photo()->create(['profile_id' => $local->profile_id]); + $bob = replyRemoteProfile(); + $mallory = replyRemoteProfile('remote.example', 'mallory'); + replySeedHosts(); + + $note = replyNote($bob, '14', $parent->url()); + replySeedFetch($note['id'], false); + + replyDeliver($mallory, $note); + + expect(Status::whereObjectUrl($note['id'])->exists())->toBeFalse(); + }); + + it('does not store a delivered object whose id is on another host', function () { + $local = replyLocalUser(); + $parent = Status::factory()->photo()->create(['profile_id' => $local->profile_id]); + $bob = replyRemoteProfile(); + replySeedHosts(); + + $note = replyNote($bob, '15', $parent->url(), [ + 'id' => 'https://other.example/users/alice/statuses/squat', + 'url' => 'https://other.example/@alice/squat', + ]); + replySeedFetch($note['id'], false); + + replyDeliver($bob, $note); + + expect(Status::whereObjectUrl($note['id'])->exists())->toBeFalse(); + }); + + it('never stores a poll vote as a comment', function () { + $local = replyLocalUser(); + $parent = Status::factory()->create(['profile_id' => $local->profile_id, 'type' => 'poll']); + $bob = replyRemoteProfile(); + replySeedHosts(); + + $vote = replyNote($bob, '16', $parent->url(), ['name' => 'Option A']); + unset($vote['content']); + + replyDeliver($bob, $vote); + + expect(Status::whereObjectUrl($vote['id'])->exists())->toBeFalse(); + }); +}); + +describe('create hardening', function () { + it('counts a vote addressed only to the poll author instead of treating it as a dm', function () { + $local = replyLocalUser(); + $bob = replyRemoteProfile(); + replySeedHosts(); + + $status = Status::factory()->create(['profile_id' => $local->profile_id, 'type' => 'poll']); + + $poll = new Poll; + $poll->status_id = $status->id; + $poll->profile_id = $local->profile_id; + $poll->poll_options = ['Yes', 'No']; + $poll->cached_tallies = [0, 0]; + $poll->votes_count = 0; + $poll->expires_at = now()->addDay(); + $poll->save(); + + // Mastodon's vote shape: to the poll author alone, no cc, no content. + $vote = replyNote($bob, 'vote', $status->url(), [ + 'name' => 'No', + 'to' => [$local->profile->permalink()], + 'cc' => [], + ]); + unset($vote['content']); + + replyDeliver($bob, $vote); + + expect($poll->fresh()->cached_tallies)->toBe([0, 1]); + expect(DirectMessage::count())->toBe(0); + expect(Status::whereObjectUrl($vote['id'])->exists())->toBeFalse(); + }); + + it('does not store a delivered post whose id is on another host', function () { + $bob = replyRemoteProfile(); + replySeedHosts(); + config(['federation.activitypub.ingest.store_notes_without_followers' => true]); + + $note = replyNote($bob, 'squat', null, [ + 'id' => 'https://other.example/users/alice/statuses/real', + 'url' => 'https://other.example/@alice/real', + 'attachment' => [[ + 'type' => 'Document', + 'mediaType' => 'image/jpeg', + 'url' => 'https://remote.example/media/2.jpg', + ]], + ]); + replySeedFetch($note['id'], false); + + replyDeliver($bob, $note); + + expect(Status::whereObjectUrl($note['id'])->exists())->toBeFalse(); + }); + + it('still stores a delivered post from its author', function () { + $bob = replyRemoteProfile(); + replySeedHosts(); + config(['federation.activitypub.ingest.store_notes_without_followers' => true]); + + $note = replyNote($bob, 'photo', null, [ + 'attachment' => [[ + 'type' => 'Document', + 'mediaType' => 'image/jpeg', + 'url' => 'https://remote.example/media/3.jpg', + ]], + ]); + + replyDeliver($bob, $note); + + $status = Status::whereObjectUrl($note['id'])->first(); + + expect($status)->not->toBeNull(); + expect((int) $status->profile_id)->toBe((int) $bob->id); + expect($status->in_reply_to_id)->toBeNull(); + }); + + it('does not hand a status held by another profile to the loser of an insert race', function () { + $bob = replyRemoteProfile(); + $mallory = replyRemoteProfile('remote.example', 'mallory'); + replySeedHosts(); + + $note = replyNote($bob, 'held', null); + + Status::factory()->create([ + 'profile_id' => $mallory->id, + 'uri' => $note['id'], + 'url' => $note['id'], + 'object_url' => $note['id'], + 'local' => false, + ]); + + $status = Helpers::createOrUpdateStatus( + $note['url'], $bob, $note['id'], $note, $note['published'], + null, false, 'public', false + ); + + expect($status)->toBeNull(); + }); +}); + +describe('retry', function () { + it('parks a delivered reply once when the parent is unavailable', function () { + $bob = replyRemoteProfile(); + replySeedHosts(); + + $missing = 'https://other.example/users/alice/statuses/slow'; + replySeedFetch($missing, false); + $note = replyNote($bob, '17', $missing); + + replyDeliver($bob, $note); + replyDeliver($bob, $note); + + expect(Status::whereObjectUrl($note['id'])->exists())->toBeFalse(); + + Queue::assertPushed(RemoteReplyResolvePipeline::class, 1); + Queue::assertPushed( + RemoteReplyResolvePipeline::class, + fn ($job) => $job->attempt === 0 && $job->queue === 'low' && $job->object['id'] === $note['id'] + ); + }); + + it('stores the parked reply once the parent resolves', function () { + $alice = replyRemoteProfile('other.example', 'alice'); + $bob = replyRemoteProfile(); + replySeedHosts(); + + $root = replyNote($alice, 'late', null); + $note = replyNote($bob, '18', $root['id']); + + replySeedFetch($root['id'], false); + replyDeliver($bob, $note); + expect(Status::count())->toBe(0); + + replySeedFetch($root['id'], ['@context' => 'https://www.w3.org/ns/activitystreams'] + $root); + (new RemoteReplyResolvePipeline($note, (int) $bob->id, 0))->handle(); + + $parent = Status::whereObjectUrl($root['id'])->first(); + $reply = Status::whereObjectUrl($note['id'])->first(); + + expect($reply)->not->toBeNull(); + expect((int) $reply->in_reply_to_id)->toBe((int) $parent->id); + expect(Cache::has(RemoteReplyResolvePipeline::pendingKey($note['id'])))->toBeFalse(); + }); + + it('schedules the next attempt, then drops the reply after the last one', function () { + $bob = replyRemoteProfile(); + replySeedHosts(); + + $missing = 'https://other.example/users/alice/statuses/never'; + replySeedFetch($missing, false); + $note = replyNote($bob, '19', $missing); + + (new RemoteReplyResolvePipeline($note, (int) $bob->id, 0))->handle(); + + Queue::assertPushed(RemoteReplyResolvePipeline::class, fn ($job) => $job->attempt === 1); + + $last = count(RemoteReplyResolvePipeline::BACKOFF) - 1; + (new RemoteReplyResolvePipeline($note, (int) $bob->id, $last))->handle(); + + Queue::assertPushed(RemoteReplyResolvePipeline::class, 1); + expect(Status::whereObjectUrl($note['id'])->exists())->toBeFalse(); + }); + + it('waits longer than a failed fetch is cached before the first retry', function () { + expect(RemoteReplyResolvePipeline::BACKOFF[0])->toBeGreaterThan(Helpers::FETCH_NEGATIVE_TTL); + }); + + it('forgets a failed fetch well before it forgets a successful one', function () { + replySeedHosts(); + $url = 'https://other.example/users/alice/statuses/flaky'; + + Http::fake(fn () => Http::response('', 503)); + expect(Helpers::fetchFromUrl($url))->toBeFalse(); + expect(Cache::get(Helpers::fetchCacheKey($url)))->toBeFalse(); + + $this->travel(Helpers::FETCH_NEGATIVE_TTL + 1)->seconds(); + + expect(Cache::get(Helpers::fetchCacheKey($url)))->toBeNull(); + }); +}); + +describe('outbound inReplyTo', function () { + it('references a remote parent by its ActivityPub id, not its permalink', function () { + $local = replyLocalUser(); + $alice = replyRemoteProfile('other.example', 'alice'); + + $parent = Status::factory()->create([ + 'profile_id' => $alice->id, + 'uri' => 'https://other.example/@alice/111', + 'url' => 'https://other.example/@alice/111', + 'object_url' => 'https://other.example/users/alice/statuses/111', + 'local' => false, + ]); + + $reply = Status::factory()->create([ + 'profile_id' => $local->profile_id, + 'in_reply_to_id' => $parent->id, + 'in_reply_to_profile_id' => $alice->id, + ]); + + $fractal = new Fractal\Manager; + $note = $fractal->createData(new Fractal\Resource\Item($reply, new Note))->toArray()['data']; + $create = $fractal->createData(new Fractal\Resource\Item($reply, new CreateNote))->toArray()['data']; + + expect($note['inReplyTo'])->toBe('https://other.example/users/alice/statuses/111'); + expect($create['object']['inReplyTo'])->toBe('https://other.example/users/alice/statuses/111'); + }); + + it('references a local parent by its status url', function () { + $local = replyLocalUser(); + $parent = Status::factory()->photo()->create(['profile_id' => $local->profile_id]); + $reply = Status::factory()->create([ + 'profile_id' => $local->profile_id, + 'in_reply_to_id' => $parent->id, + ]); + + expect($reply->inReplyToUri())->toBe($parent->url()); + expect($parent->inReplyToUri())->toBeNull(); + }); +}); diff --git a/tests/Feature/StatusDeleteReplyCleanupTest.php b/tests/Feature/StatusDeleteReplyCleanupTest.php new file mode 100644 index 000000000..35e640c10 --- /dev/null +++ b/tests/Feature/StatusDeleteReplyCleanupTest.php @@ -0,0 +1,122 @@ + false]); +}); + +function cleanupRemoteProfile(): Profile +{ + return Profile::factory()->remote()->create([ + 'domain' => 'remote.example', + 'remote_url' => 'https://remote.example/users/bob', + ]); +} + +function cleanupRemoteReply(Profile $author, Status $parent, string $path): Status +{ + $id = "https://remote.example/users/bob/statuses/{$path}"; + + return Status::factory()->create([ + 'profile_id' => $author->id, + 'in_reply_to_id' => $parent->id, + 'in_reply_to_profile_id' => $parent->profile_id, + 'uri' => $id, + 'url' => $id, + 'object_url' => $id, + 'local' => false, + ]); +} + +it('queues remote replies for deletion when a local status is deleted', function () { + Queue::fake(); + + $user = User::factory()->create(); + $user->refresh(); + $parent = Status::factory()->photo()->create(['profile_id' => $user->profile_id]); + $reply = cleanupRemoteReply(cleanupRemoteProfile(), $parent, '1'); + + (new StatusDelete($parent))->handle(); + + Queue::assertPushedOn('delete', RemoteStatusDelete::class); + + // Still attached while it waits: hidden, never a top-level post. + expect((int) $reply->fresh()->in_reply_to_id)->toBe((int) $parent->id); +}); + +it('removes remote replies, and their remote replies, when the jobs run', function () { + $user = User::factory()->create(); + $user->refresh(); + $bob = cleanupRemoteProfile(); + + $parent = Status::factory()->photo()->create(['profile_id' => $user->profile_id]); + $reply = cleanupRemoteReply($bob, $parent, '2'); + $nested = cleanupRemoteReply($bob, $reply, '3'); + + (new StatusDelete($parent))->handle(); + + expect(Status::withTrashed()->find($reply->id))->toBeNull(); + expect(Status::withTrashed()->find($nested->id))->toBeNull(); +}); + +it('removes remote replies when a remote status is deleted', function () { + $bob = cleanupRemoteProfile(); + + $parent = Status::factory()->photo()->create([ + 'profile_id' => $bob->id, + 'uri' => 'https://remote.example/users/bob/statuses/root', + 'url' => 'https://remote.example/users/bob/statuses/root', + 'object_url' => 'https://remote.example/users/bob/statuses/root', + 'local' => false, + ]); + $reply = cleanupRemoteReply($bob, $parent, '4'); + + (new RemoteStatusDelete($parent))->handle(); + + expect(Status::withTrashed()->find($parent->id))->toBeNull(); + expect(Status::withTrashed()->find($reply->id))->toBeNull(); +}); + +it('keeps and detaches local replies', function () { + $user = User::factory()->create(); + $user->refresh(); + $commenter = User::factory()->create(); + $commenter->refresh(); + + $parent = Status::factory()->photo()->create(['profile_id' => $user->profile_id]); + $comment = Status::factory()->create([ + 'profile_id' => $commenter->profile_id, + 'in_reply_to_id' => $parent->id, + 'in_reply_to_profile_id' => $user->profile_id, + ]); + + (new StatusDelete($parent))->handle(); + + expect($comment->fresh())->not->toBeNull(); + expect($comment->fresh()->in_reply_to_id)->toBeNull(); +});