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
3 changes: 3 additions & 0 deletions database/factories/MemberFactory.php
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,9 @@ public function definition()
return [
'channel_id' => Channel::factory(),
'socket_id' => fake()->uuid,
'user_id' => fn (array $attributes) => isset($attributes['data']['user_id'])
? (string) $attributes['data']['user_id']
: null,
'data' => null,
];
}
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,29 @@
<?php declare(strict_types=1);

use Illuminate\Database\Migrations\Migration;
use Illuminate\Database\Schema\Blueprint;
use Illuminate\Support\Facades\Schema;

return new class extends Migration
{
/** @var string */
private $table = 'pollcast_channel_members';

public function up(): void
{
Schema::table($this->table, function (Blueprint $table) {
$table->string('user_id', 191)->nullable()->after('socket_id');
$table->index(['channel_id', 'user_id']);
});
}

public function down(): void
{
Schema::table($this->table, function (Blueprint $table) {
$table->dropForeign(['channel_id']);
$table->dropIndex(['channel_id', 'user_id']);
$table->dropColumn('user_id');
$table->foreign('channel_id')->references('id')->on('pollcast_channel')->cascadeOnDelete();
});
}
};
52 changes: 50 additions & 2 deletions src/Broadcasting/Socket.php
Original file line number Diff line number Diff line change
Expand Up @@ -176,7 +176,10 @@ public function joinChannel(string $name, ?array $data = null): void
$member = Member::query()->firstOrCreate([
'channel_id' => $channel->id,
'socket_id' => $this->getId(),
], ['data' => $data]);
], [
'user_id' => isset($data['user_id']) ? (string) $data['user_id'] : null,
'data' => $data,
]);
$member->touch();

if ($data === null) {
Expand All @@ -194,6 +197,10 @@ public function removeMemberFromChannel(Member $member, Channel $channel): void
return;
}

if ($this->hasAnotherSocket($member)) {
return;
}

(new Message([
'channel_id' => $channel->id,
// The member's own socket, not the request's - garbage collection removes members
Expand All @@ -215,9 +222,19 @@ protected function joinedPresenceChannel(Channel $channel, Member $member, array
'channel_id' => $channel->id,
'member_id' => $member->id,
'event' => 'pollcast:subscription_succeeded',
'payload' => Member::query()->where('channel_id', $channel->id)->pluck('data'),
'payload' => Member::query()
->where('channel_id', $channel->id)
->get()
->unique(fn (Member $m) => $m->user_id ?? $m->socket_id)
->pluck('data')
->values(),
]))->save();

// A user is announced once, however many sockets they have on the channel.
if (! $this->isEarliestSocket($member)) {
return;
}

// Broadcast member added event to everyone else in the channel - the joiner already
// has themselves in the subscription_succeeded member list.
(new Message([
Expand All @@ -228,6 +245,37 @@ protected function joinedPresenceChannel(Channel $channel, Member $member, array
]))->save();
}

private function hasAnotherSocket(Member $member): bool
{
if ($member->user_id === null) {
return false;
}

return Member::query()
->where('channel_id', $member->channel_id)
->where('user_id', $member->user_id)
->where('id', '!=', $member->id)
->exists();
}

/**
* Whether this is the user's earliest socket on the channel, which is the one that announces them. Unlike
* checking for any other socket, two sockets which join at the same time cannot both decide not to.
*/
private function isEarliestSocket(Member $member): bool
{
if ($member->user_id === null) {
return true;
}

return Member::query()
->where('channel_id', $member->channel_id)
->where('user_id', $member->user_id)
->orderBy('created_at')
->orderBy('id')
->value('id') === $member->id;
}

/**
* @throws ExpiredException
* @throws InvalidSocketException
Expand Down
4 changes: 3 additions & 1 deletion src/Model/Member.php
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
* @property-read string $id
* @property-read string $channel_id
* @property-read string $socket_id
* @property-read string|null $user_id
* @property-read mixed[] $data
*/
class Member extends Model
Expand All @@ -33,12 +34,13 @@ class Member extends Model
protected $guarded = [];

/** @var list<string> */
protected $fillable = ['channel_id', 'socket_id', 'data'];
protected $fillable = ['channel_id', 'socket_id', 'user_id', 'data'];

/** @var array<string, string> */
protected $casts = [
'channel_id' => 'string',
'socket_id' => 'string',
'user_id' => 'string',
'data' => 'json',
];

Expand Down
53 changes: 53 additions & 0 deletions tests/Unit/SocketTest.php
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
use SupportPal\Pollcast\Exception\InvalidSocketException;
use SupportPal\Pollcast\Model\Channel;
use SupportPal\Pollcast\Model\Member;
use SupportPal\Pollcast\Model\Message;
use SupportPal\Pollcast\Tests\TestCase;

use function app;
Expand Down Expand Up @@ -134,6 +135,7 @@ public function testJoinPresenceChannel(): void
$this->assertDatabaseHas('pollcast_channel_members', [
'channel_id' => $channel->id,
'socket_id' => $socketId,
'user_id' => '1',
'data' => json_encode($data),
]);

Expand All @@ -156,6 +158,42 @@ public function testJoinPresenceChannel(): void
]);
}

public function testJoinPresenceChannelWithAnotherSocketOfTheSameUser(): void
{
$socket = new Socket(app('config'), app('session.store'), request());
$socket->setId('test');

$data = ['user_id' => 1, 'user_info' => ['name' => 'Someone']];
$channel = Channel::factory()->create(['name' => 'presence-channel']);
Member::factory()->create(['channel_id' => $channel->id, 'socket_id' => 'other-tab', 'data' => $data, 'created_at' => now()->subMinute()]);
Member::factory()->create(['channel_id' => $channel->id, 'socket_id' => 'someone-else', 'data' => $other = ['user_id' => 2]]);

$socket->joinChannel($channel->name, $data);

$this->assertDatabaseHas('pollcast_message_queue', [
'channel_id' => $channel->id,
'event' => 'pollcast:subscription_succeeded',
'payload' => json_encode([$data, $other]),
]);
$this->assertDatabaseMissing('pollcast_message_queue', ['event' => 'pollcast:member_added']);
}

public function testJoinPresenceChannelAtTheSameTimeAsAnotherSocketOfTheSameUser(): void
{
$data = ['user_id' => 1, 'user_info' => ['name' => 'Someone']];
$channel = Channel::factory()->create(['name' => 'presence-channel']);

// Both sockets' rows exist before either has checked for the other.
$tabA = Member::factory()->create(['channel_id' => $channel->id, 'socket_id' => 'tab-a', 'data' => $data]);
Member::factory()->create(['channel_id' => $channel->id, 'socket_id' => 'tab-b', 'data' => $data, 'created_at' => $tabA->created_at]);

foreach (['tab-a', 'tab-b'] as $socketId) {
(new Socket(app('config'), app('session.store'), request()))->setId($socketId)->joinChannel($channel->name, $data);
}

$this->assertSame(1, Message::query()->where('event', 'pollcast:member_added')->count());
}

public function testRemoveMemberFromChannel(): void
{
$socketId = 'test';
Expand Down Expand Up @@ -198,6 +236,21 @@ public function testRemoveMemberFromPresenceChannel(): void
]);
}

public function testRemoveMemberFromPresenceChannelWithAnotherSocketOfTheSameUser(): void
{
$socket = new Socket(app('config'), app('session.store'), request());

$data = ['user_id' => 1];
$channel = Channel::factory()->create(['name' => 'presence-name']);
$member = Member::factory()->create(['channel_id' => $channel->id, 'data' => $data]);
Member::factory()->create(['channel_id' => $channel->id, 'data' => $data]);

$socket->removeMemberFromChannel($member, $channel);

$this->assertDatabaseMissing('pollcast_channel_members', ['id' => $member->id]);
$this->assertDatabaseMissing('pollcast_message_queue', ['event' => 'pollcast:member_removed']);
}

/**
* A private channel is authorised like a presence one but publishes nothing about who is on
* it, so a member leaving is not announced to the others.
Expand Down
Loading