Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 14 additions & 0 deletions app/Dto/Engagement/ReplyPostResult.php
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ private function __construct(
public ?string $remoteReplyId = null,
public ?string $remoteCid = null,
public ?string $message = null,
public ?string $processingContainerId = null,
) {}

public static function ok(string $remoteReplyId, ?string $remoteCid = null): self
Expand Down Expand Up @@ -40,6 +41,19 @@ public static function failed(?string $message = null): self
return new self(EngagementStatus::Failed, message: $message);
}

/**
* The platform has not finished the reply. Anything that cannot wait for it
* sees an ordinary failure.
*/
public static function processing(string $containerId): self
{
return new self(
EngagementStatus::Failed,
message: 'The platform is still processing the reply. Try again in a moment.',
processingContainerId: $containerId,
);
}

public function isOk(): bool
{
return $this->status->isOk();
Expand Down
9 changes: 9 additions & 0 deletions app/Enums/Platform.php
Original file line number Diff line number Diff line change
Expand Up @@ -156,6 +156,15 @@ public function supportsReplyMedia(): bool
};
}

/**
* Threads builds a reply in a container that takes time to finish, so its
* replies always go through SendReply and never wait in a web request.
*/
public function sendsRepliesInBackground(): bool
{
return $this === self::Threads;
}

/**
* Whether this platform's metrics connector returns real post-level metrics.
* Every launched platform, including LinkedIn (Page/Organization accounts
Expand Down
22 changes: 22 additions & 0 deletions app/Exceptions/ReplyMayHavePosted.php
Original file line number Diff line number Diff line change
@@ -0,0 +1,22 @@
<?php

declare(strict_types=1);

namespace App\Exceptions;

use RuntimeException;
use Throwable;

/**
* The platform accepted the reply but we could not record it. A retry would
* post it twice, so the job fails at once and tells the user to check first.
*/
class ReplyMayHavePosted extends RuntimeException
{
public const string USER_MESSAGE = 'The reply may have posted. Check the post before you send it again.';

public function __construct(Throwable $previous)
{
parent::__construct('Reply posted but not recorded: '.$previous->getMessage(), previous: $previous);
}
}
12 changes: 10 additions & 2 deletions app/Http/Controllers/Engagement/EngagementController.php
Original file line number Diff line number Diff line change
Expand Up @@ -357,7 +357,15 @@ public function respond(

$mediaIds = array_values($request->validated('media', []));

if ($mediaIds === []) {
// A new reply replaces our failed attempts at this one; they never reached the platform.
PostTargetReply::withoutGlobalScopes()
->where('post_target_id', $reply->post_target_id)
->where('parent_remote_id', $reply->remote_reply_id)
->where('is_ours', true)
->where('send_status', SendStatus::Failed->value)
->delete();

if ($mediaIds === [] && ! $reply->platform->sendsRepliesInBackground()) {
$result = $registry->for($reply->platform)->postReply($account, $reply, (string) $request->validated('text'), $credentials);

if (! $result->isOk()) {
Expand Down Expand Up @@ -393,7 +401,7 @@ public function respond(
return $this->respondedRow($ourRow);
}

// Media path: create the outgoing row in a sending state, then dispatch.
// Background path: create the outgoing row in a sending state, then dispatch.
$ourRow = PostTargetReply::withoutGlobalScopes()->create([
'workspace_id' => $reply->workspace_id,
'post_target_id' => $reply->post_target_id,
Expand Down
2 changes: 1 addition & 1 deletion app/Http/Controllers/Posts/PostQueueController.php
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,7 @@ public function store(Request $request, Post $post, WorkspaceSubscriptionGate $s
'message' => $request->filled('scheduled_at')
? 'Choose an open slot from your posting queue.'
: 'No open posting slot available. Add posting-schedule slots in settings.',
], 422);
], 409);
}

$post->scheduled_at = $slot;
Expand Down
4 changes: 2 additions & 2 deletions app/Http/Controllers/Posts/PublishController.php
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@ public function store(
if ($post->loadMissing('targets')->targets->isEmpty()) {
return response()->json([
'message' => 'Select at least one account to publish.',
], 422);
], 409);
}

if (! $subscriptions->canPublish($workspace)) {
Expand All @@ -45,7 +45,7 @@ public function store(
return response()->json([
'message' => "Some accounts can't be published yet.",
'blocked' => $blocked,
], 422);
], 409);
}

$post->forceFill(['status' => PostStatus::Publishing->value])->save();
Expand Down
2 changes: 2 additions & 0 deletions app/Http/Controllers/WorkspaceMentionController.php
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,8 @@ public function store(Request $request): JsonResponse
'name' => ['required', 'string', 'max:80', 'regex:/^@[A-Za-z0-9_.-]+$/'],
'handles' => ['required', 'array'],
...$handleRules,
], [
'name.regex' => 'A mention name can use only letters A to Z, numbers, dots, dashes, and underscores.',
]);
$handles = [];
foreach ($allowedPlatforms as $platform) {
Expand Down
3 changes: 2 additions & 1 deletion app/Jobs/DeletePostTarget.php
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
use App\Models\PostTarget;
use App\Services\Publishing\PublishConnectorRegistry;
use App\Services\Publishing\TokenManager;
use App\Support\ConnectionFailure;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;
use Illuminate\Support\Str;
Expand Down Expand Up @@ -54,7 +55,7 @@ public function failed(Throwable $e): void

$target->forceFill([
'status' => PostTargetStatus::Failed->value,
'error_message' => Str::limit($e->getMessage(), 1000),
'error_message' => Str::limit(ConnectionFailure::safeMessage($e), 1000),
'next_attempt_at' => null,
])->save();
}
Expand Down
9 changes: 6 additions & 3 deletions app/Jobs/PublishPostTarget.php
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
use App\Services\Publishing\PublishConnectorRegistry;
use App\Services\Publishing\SegmentMediaResolver;
use App\Services\Publishing\TokenManager;
use App\Support\ConnectionFailure;
use App\Support\InstanceSettings;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;
Expand Down Expand Up @@ -172,7 +173,7 @@ public function handle(
// A transient token-endpoint failure (429/5xx/timeout) is not a bad
// credential — treat it as a retryable server error so the publish backs
// off and retries instead of flipping the account to needs-attention.
$result = PublishResult::failure(ErrorKind::ServerError, $e->getMessage());
$result = PublishResult::failure(ErrorKind::ServerError, ConnectionFailure::safeMessage($e));
} catch (TokenRefreshException $e) {
$result = PublishResult::failure(ErrorKind::AuthExpired, $e->getMessage());
}
Expand Down Expand Up @@ -292,11 +293,13 @@ private function resumePartial(PostTarget $target): void

private function markFailed(PostTarget $target, Throwable $e): void
{
$this->closeOpenAttempt($target, 'failed', Str::limit($e->getMessage(), 1000));
$message = Str::limit(ConnectionFailure::safeMessage($e), 1000);

$this->closeOpenAttempt($target, 'failed', $message);

$target->forceFill([
'status' => PostTargetStatus::Failed->value,
'error_message' => Str::limit($e->getMessage(), 1000),
'error_message' => $message,
'next_attempt_at' => null,
])->save();

Expand Down
78 changes: 61 additions & 17 deletions app/Jobs/SendReply.php
Original file line number Diff line number Diff line change
Expand Up @@ -7,12 +7,15 @@
use App\Enums\Platform;
use App\Enums\ReplyStatus;
use App\Enums\SendStatus;
use App\Exceptions\ReplyMayHavePosted;
use App\Exceptions\TokenRefreshException;
use App\Models\PostMedia;
use App\Models\PostTargetReply;
use App\Notifications\ReplyFailedNotification;
use App\Services\Engagement\Contracts\PublishesReplyContainers;
use App\Services\Engagement\EngagementConnectorRegistry;
use App\Services\Publishing\TokenManager;
use DateTimeInterface;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;
use Illuminate\Queue\Middleware\RateLimited;
Expand All @@ -22,11 +25,22 @@ class SendReply implements ShouldQueue
{
use Queueable;

public int $tries = 3;
public const int CONTAINER_CHECK_DELAY_SECONDS = 3;

public const int MAX_CONTAINER_CHECKS = 20;

/**
* A rate-limit release still counts as an attempt, so a reply waiting for its
* turn would fail after a few releases. Retry by time, and stop on real errors.
*/
public int $maxExceptions = 3;

public int $timeout = 300;

/**
* A processing container comes back as a follow-up job, so the wait never
* holds a queue worker. `$containerId` keeps it from creating a second one.
*
* @param list<string> $mediaIds
*/
public function __construct(
Expand All @@ -35,17 +49,25 @@ public function __construct(
public array $mediaIds,
public string $text,
public Platform $platform,
public ?string $containerId = null,
public int $containerChecks = 0,
) {}

/**
* Throttle outbound replies on the same per-platform limiter as the fetch job,
* so a burst of replies can't trip the platform's own rate limits.
* so a burst of replies can't trip the platform's own rate limits. A follow-up
* check is not a new send, and would otherwise spend the limit on status reads.
*
* @return array<int, object>
*/
public function middleware(): array
{
return [new RateLimited("engagement-{$this->platform->value}")];
return $this->containerId === null ? [new RateLimited("engagement-{$this->platform->value}")] : [];
}

public function retryUntil(): DateTimeInterface
{
return now()->addMinutes(10);
}

/**
Expand Down Expand Up @@ -79,7 +101,7 @@ public function handle(EngagementConnectorRegistry $registry, TokenManager $toke
}

try {
$credentials = in_array($account->platform, [Platform::X, Platform::Bluesky, Platform::LinkedIn], true)
$credentials = in_array($account->platform, [Platform::X, Platform::Bluesky, Platform::LinkedIn, Platform::Threads], true)
? $tokens->fresh($account)
: [];
} catch (TokenRefreshException) {
Expand All @@ -90,7 +112,24 @@ public function handle(EngagementConnectorRegistry $registry, TokenManager $toke

$media = array_values(PostMedia::withoutGlobalScopes()->whereIn('id', $this->mediaIds)->get()->all());

$result = $registry->for($parent->platform)->postReply($account, $parent, $this->text, $credentials, $media);
$connector = $registry->for($parent->platform);
$result = $this->containerId !== null && $connector instanceof PublishesReplyContainers
? $connector->publishContainer($account, $this->containerId, $credentials)
: $connector->postReply($account, $parent, $this->text, $credentials, $media);
Comment on lines +116 to +118

if ($result->processingContainerId !== null && $this->containerChecks < self::MAX_CONTAINER_CHECKS) {
self::dispatch(
$this->ourRowId,
$this->parentReplyId,
$this->mediaIds,
$this->text,
$this->platform,
$result->processingContainerId,
$this->containerChecks + 1,
)->delay(self::CONTAINER_CHECK_DELAY_SECONDS);

return;
}

if (! $result->isOk()) {
// The connector's own words — e.g. "LinkedIn comments cannot include
Expand All @@ -101,16 +140,21 @@ public function handle(EngagementConnectorRegistry $registry, TokenManager $toke
return;
}

$ourRow->forceFill([
'remote_reply_id' => $result->remoteReplyId,
'remote_cid' => $result->remoteCid,
'send_status' => SendStatus::Sent->value,
])->save();

$parent->forceFill([
'status' => ReplyStatus::Responded->value,
'our_reply_remote_id' => $result->remoteReplyId,
])->save();
// The platform has the reply now: a retry from here would post it twice.
try {
$ourRow->forceFill([
'remote_reply_id' => $result->remoteReplyId,
'remote_cid' => $result->remoteCid,
'send_status' => SendStatus::Sent->value,
])->save();

$parent->forceFill([
'status' => ReplyStatus::Responded->value,
'our_reply_remote_id' => $result->remoteReplyId,
])->save();
} catch (Throwable $e) {
$this->fail(new ReplyMayHavePosted($e));
}
}

/**
Expand All @@ -125,13 +169,13 @@ public function failed(Throwable $e): void
// Don't report failure (or re-notify) for a reply that actually went out
// — handle() may have posted successfully and thrown only afterwards.
if ($ourRow !== null && $ourRow->send_status !== SendStatus::Sent) {
$this->failRow($ourRow);
$this->failRow($ourRow, $e instanceof ReplyMayHavePosted ? ReplyMayHavePosted::USER_MESSAGE : null);
}
}

private function failRow(PostTargetReply $ourRow, ?string $reason = null): void
{
$ourRow->forceFill(['send_status' => SendStatus::Failed->value])->save();
$ourRow->forceFill(['send_status' => SendStatus::Failed->value, 'send_error' => $reason])->save();

$author = $ourRow->target?->post()->withoutGlobalScopes()->first()?->author()->first();
$parent = PostTargetReply::withoutGlobalScopes()->find($this->parentReplyId);
Expand Down
2 changes: 2 additions & 0 deletions app/Models/PostTargetReply.php
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@
* @property string|null $like_remote_id
* @property bool $is_ours
* @property SendStatus|null $send_status
* @property string|null $send_error
* @property CarbonImmutable $fetched_at
*/
#[Fillable([
Expand All @@ -60,6 +61,7 @@
'like_remote_id',
'is_ours',
'send_status',
'send_error',
'fetched_at',
])]
class PostTargetReply extends Model
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
use App\Services\Atproto\DPoP;
use App\Services\Engagement\Contracts\EngagementConnector;
use App\Services\Usage\Concerns\TracksUsage;
use App\Support\ConnectionFailure;
use App\Support\RetryAfter;
use App\Support\UsageOperation;
use Carbon\CarbonImmutable;
Expand Down Expand Up @@ -216,7 +217,7 @@ public function postReply(ConnectedAccount $account, PostTargetReply $parent, st
} catch (BlueskyReplyMediaFailed) {
return ReplyPostResult::failed('Could not upload media.');
} catch (ConnectionException $e) {
return ReplyPostResult::failed($e->getMessage());
return ReplyPostResult::failed(ConnectionFailure::report($e));
}

$this->meter(UsageCategory::ExternalApi, UsageOperation::REPLY_SEND, $account, $response);
Expand Down Expand Up @@ -245,7 +246,7 @@ public function likeReply(ConnectedAccount $account, PostTargetReply $reply, arr
],
]);
} catch (ConnectionException $e) {
return ReplyActionResult::failed($e->getMessage());
return ReplyActionResult::failed(ConnectionFailure::report($e));
}

$this->meter(UsageCategory::ExternalApi, UsageOperation::REPLY_LIKE, $account, $response);
Expand Down Expand Up @@ -293,7 +294,7 @@ private function deleteRecord(ConnectedAccount $account, array $credentials, str
'rkey' => $rkey,
]);
} catch (ConnectionException $e) {
return ReplyActionResult::failed($e->getMessage());
return ReplyActionResult::failed(ConnectionFailure::report($e));
}

$this->meter(UsageCategory::ExternalApi, $operation, $account, $response);
Expand Down
Loading
Loading