Repository: supabase-community/supabase-kt (kotlin)

Issue:
[Bug]: Realtime - Subscribing to the same channel id causes upstream closes / unexpected behavior.
### General Info

- [X] I checked for similar bug report
- [ ] I am using the latest version (latest/3.0.3 doesn't seem to solve to issue)
- [X] I checked the [troubleshooting](https://github.com/supabase-community/supabase-kt/blob/master/TROUBLESHOOTING.md) page for similar problems

### Version(s)

3.0.2

### Kotlin Target(s) and their respective versions

2.0.21 / JVM 8

### What happened? (include your code)

Hey everyone, love the work everyone has done.

I am not sure if I missed some details in the documentation here, or I am misunderstanding how presence works, or this is actually a bug. I tried tinkering around with channels, and tried connecting multiple separate coroutines ("user clients") to some channel, each client represents some user who is interested in the channel (think microservices). I expect it to just work, but it seems when the second user subscribes to the channel, the upstream is sending a close command to the first subscriber. If I understand correctly, channels do not have to be created ahead of time. Given the upstream server seems to be forcing the channel closed, I am a bit confused why. I dug into the Realtime logs, and the only hint I see is: **"Tenant has no connected users, database connection will be terminated"**. Am I restricted from doing some in-process communication on the same channel? (1 channel per client) Do I need to be cautious of where the channel is sourcing from, and avoid subscribing to the channel if it happened to be on the same JVM/Client?

Something to note: If they all subscribe to different channels (ex: `main-$user` instead of `main`), it works fine. 
It doesn't seem to be permissions or quota.

**Edit**: Seems to be indeed 1 channel subscription (by id) per connection. Don't see where this is documented. If this is confirmed the case, perhaps the behavior should be to throw an error when it detects a existing subscription?

Thank you for your time.


```kotlin

import io.github.jan.supabase.createSupabaseClient
import io.github.jan.supabase.logging.LogLevel
import io.github.jan.supabase.realtime.Realtime
import io.github.jan.supabase.realtime.channel
import io.github.jan.supabase.realtime.presenceDataFlow
import io.github.jan.supabase.realtime.realtime
import io.github.jan.supabase.realtime.track
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.flow.launchIn
import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.job
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlinx.serialization.Serializable

object Realtime {

    @Serializable
    data class Pres(val data: String)

    @JvmStatic
    fun main(args: Array<String>): Unit = runBlocking(Dispatchers.Default) {

        val client = createSupabaseClient(
            supabaseUrl = "<REDACTED>",
            supabaseKey = "<REDACTED>"
        ) {
            install(Realtime)
            defaultLogLevel = LogLevel.DEBUG
        }

        val realtime = client.realtime

        realtime.status.onEach {
            println("realtime.status = $it")
        }.launchIn(this)

        realtime.connect()

        repeat(3) { id ->
            launchClient(realtime, "user-$id")
        }

    }

    fun CoroutineScope.launchClient(realtime: Realtime, user: String) {
        launch {

            // all "user clients" attempt to connect to some common channel
            val channel = realtime.channel("main") {
                presence { key = user }
            }

            fun log(msg: Any?) = println("$user: $msg")

            channel.status.onEach { status ->
                log("status == $status")
            }.launchIn(this)

            channel.presenceDataFlow<Pres>().onEach { data ->
                log("data == $data")
            }.launchIn(this)

            log("subscribing...")
            channel.subscribe(blockUntilSubscribed = true)

            log("tracking...")
            channel.track(Pres("$user is ready"))

            log("waiting...")
            coroutineContext.job.join()
        }

    }

}
```

### Steps To Reproduce (optional)

Run the snippet and you should see none of the "clients" make it past the tracking step.

### Relevant log output (optional)

```shell
Connected to the target VM, address: '127.0.0.1:58569', transport: 'socket'
Info: (Supabase-Core) SupabaseClient created! Please report any bugs you find.
realtime.status = CONNECTING
Info: (Supabase-Realtime) Connected to realtime websocket!
realtime.status = CONNECTED
user-0: status == UNSUBSCRIBED
user-2: status == UNSUBSCRIBED
user-1: status == UNSUBSCRIBED
user-2: subscribing...
user-0: subscribing...
user-1: subscribing...
Debug: (Supabase-Realtime) Subscribing to channel realtime:main
user-0: status == SUBSCRIBING
Debug: (Supabase-Realtime) Subscribing to channel realtime:main
Debug: (Supabase-Realtime) Subscribing to channel realtime:main
user-1: status == SUBSCRIBING
user-2: status == SUBSCRIBING
Debug: (Supabase-Realtime) Subscribing to channel with body {"config":{"broadcast":{"ack":false,"self":false},"presence":{"key":"user-0"},"postgres_changes":[],"private":false}}
Debug: (Supabase-Realtime) Subscribing to channel with body {"config":{"broadcast":{"ack":false,"self":false},"presence":{"key":"user-1"},"postgres_changes":[],"private":false}}
Debug: (Supabase-Realtime) Subscribing to channel with body {"config":{"broadcast":{"ack":false,"self":false},"presence":{"key":"user-2"},"postgres_changes":[],"private":false}}
Debug: (Supabase-Realtime) Received message RealtimeMessage(topic=realtime:main, event=phx_reply, payload={"status":"ok","response":{"postgres_changes":[]}}, ref=null)
Debug: (Supabase-Realtime) Received event phx_reply for channel realtime:main
Debug: (Supabase-Realtime) Joined channel realtime:main
user-2: status == SUBSCRIBED
Debug: (Supabase-Realtime) Received message RealtimeMessage(topic=realtime:main, event=phx_reply, payload={"status":"ok","response":{"postgres_changes":[]}}, ref=null)
Debug: (Supabase-Realtime) Received event phx_reply for channel realtime:main
Debug: (Supabase-Realtime) Received message RealtimeMessage(topic=realtime:main, event=phx_reply, payload={"status":"ok","response":{"postgres_changes":[]}}, ref=null)
Debug: (Supabase-Realtime) Received event phx_reply for channel realtime:main
Debug: (Supabase-Realtime) Received message RealtimeMessage(topic=realtime:main, event=presence_state, payload={}, ref=null)
Debug: (Supabase-Realtime) Received event presence_state for channel realtime:main
Debug: (Supabase-Realtime) Received message RealtimeMessage(topic=realtime:main, event=phx_close, payload={}, ref=null)
Debug: (Supabase-Realtime) Received event phx_close for channel realtime:main
user-2: tracking...
Debug: (Supabase-Realtime) Unsubscribing from channel realtime:main
user-2: status == UNSUBSCRIBING
Debug: (Supabase-Realtime) No more subscriptions, disconnecting from realtime websocket
Debug: (Supabase-Realtime) Closing websocket connection
user-2: data == []
Debug: (Supabase-Realtime) Unsubscribed from channel realtime:main
realtime.status = DISCONNECTED
Exception in thread "main" java.lang.IllegalStateException: You can only track your presence after subscribing to the channel. Did you forget to call `channel.subscribe()`?
	at io.github.jan.supabase.realtime.RealtimeChannelImpl.track(RealtimeChannelImpl.kt:140)
	at mystic.proxyinspector.Realtime$launchClient$1.invokeSuspend(Realtime.kt:89)
	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith$$$capture(ContinuationImpl.kt:33)
	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt)
	at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:101)
	at kotlinx.coroutines.scheduling.CoroutineScheduler.runSafely(CoroutineScheduler.kt:589)
	at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.executeTask(CoroutineScheduler.kt:832)
	at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.runWorker(CoroutineScheduler.kt:720)
	at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.run(CoroutineScheduler.kt:707)
Disconnected from the target VM, address: '127.0.0.1:58569', transport: 'socket'

Process finished with exit code 1
```

Pull request description:
## What kind of change does this PR introduce?

Docs update / Improvement (closes #818)

## What is the current behavior?

Calling `Realtime#channel()` will always create a new channel and override the existing one, if there is one by the given channel id. Also the docs are not very clear.

## What is the new behavior?

- *`Realtime#channel()` will now also return an existing channel, if there is one.*
- The internal implementation and the docs have also been improved.

Patch:
diff --git a/Realtime/src/commonMain/kotlin/io/github/jan/supabase/realtime/Realtime.kt b/Realtime/src/commonMain/kotlin/io/github/jan/supabase/realtime/Realtime.kt
index 774d6bf60..20f665714 100644
--- a/Realtime/src/commonMain/kotlin/io/github/jan/supabase/realtime/Realtime.kt
+++ b/Realtime/src/commonMain/kotlin/io/github/jan/supabase/realtime/Realtime.kt
@@ -68,9 +68,6 @@ sealed interface Realtime : MainPlugin<Realtime.Config>, CustomSerializationPlug
      */
     fun disconnect()
 
-    @SupabaseInternal
-    fun Realtime.addChannel(channel: RealtimeChannel)
-
     /**
      * Unsubscribes and removes a channel from the [subscriptions]
      * @param channel The channel to remove
@@ -103,6 +100,17 @@ sealed interface Realtime : MainPlugin<Realtime.Config>, CustomSerializationPlug
      */
     suspend fun setAuth(token: String? = null)
 
+    /**
+     * Creates a new [RealtimeChannel] and adds it to the [subscriptions]
+     *
+     * - This method does not subscribe to the channel. You have to call [RealtimeChannel.subscribe] to do so.
+     * - If a channel with the same [channelId] already exists, it will be returned
+     *
+     * @param channelId The id of the channel
+     * @param builder The builder for the channel
+     */
+    fun channel(channelId: String, builder: RealtimeChannelBuilder): RealtimeChannel
+
     /**
      * @property websocketConfig Custom configuration for the Ktor Websocket Client. This only applies if [Realtime.Config.websocketFactory] is null.
      * @property secure Whether to use wss or ws. Defaults to [SupabaseClient.useHTTPS] when null
@@ -187,11 +195,15 @@ sealed interface Realtime : MainPlugin<Realtime.Config>, CustomSerializationPlug
 }
 
 /**
- * Creates a new [RealtimeChannel]
+ * Creates a new [RealtimeChannel] and adds it to the [Realtime.subscriptions]
+ *
+ * - This method does not subscribe to the channel. You have to call [RealtimeChannel.subscribe] to do so.
+ * - If a channel with the same [channelId] already exists, it will be returned
+ *
+ * @param channelId The id of the channel
+ * @param builder The builder for the channel
  */
-inline fun Realtime.channel(channelId: String, builder: RealtimeChannelBuilder.() -> Unit = {}): RealtimeChannel {
-    return RealtimeChannelBuilder("realtime:$channelId", this as RealtimeImpl).apply(builder).build()
-}
+inline fun Realtime.channel(channelId: String, builder: RealtimeChannelBuilder.() -> Unit = {}): RealtimeChannel = channel(channelId, RealtimeChannelBuilder(RealtimeTopic.withChannelId(channelId)).apply(builder))
 
 /**
  * Supabase Realtime is a way to listen to changes in the PostgreSQL database via websockets
diff --git a/Realtime/src/commonMain/kotlin/io/github/jan/supabase/realtime/RealtimeChannelBuilder.kt b/Realtime/src/commonMain/kotlin/io/github/jan/supabase/realtime/RealtimeChannelBuilder.kt
index 14e6d954f..1edd17c35 100644
--- a/Realtime/src/commonMain/kotlin/io/github/jan/supabase/realtime/RealtimeChannelBuilder.kt
+++ b/Realtime/src/commonMain/kotlin/io/github/jan/supabase/realtime/RealtimeChannelBuilder.kt
@@ -7,7 +7,7 @@ import io.github.jan.supabase.realtime.annotations.ChannelDsl
  * Used to build a realtime channel
  */
 @ChannelDsl
-class RealtimeChannelBuilder @PublishedApi internal constructor(private val topic: String, private val realtimeImpl: RealtimeImpl) {
+class RealtimeChannelBuilder @PublishedApi internal constructor(private val topic: String) {
 
     private var broadcastJoinConfig = BroadcastJoinConfig(acknowledgeBroadcasts = false, receiveOwnBroadcasts = false)
     private var presenceJoinConfig = PresenceJoinConfig("")
@@ -33,9 +33,9 @@ class RealtimeChannelBuilder @PublishedApi internal constructor(private val topi
     }
 
     @SupabaseInternal
-    fun build(): RealtimeChannel {
+    fun build(realtime: Realtime): RealtimeChannel {
         return RealtimeChannelImpl(
-            realtimeImpl,
+            realtime,
             topic,
             broadcastJoinConfig,
             presenceJoinConfig,
diff --git a/Realtime/src/commonMain/kotlin/io/github/jan/supabase/realtime/RealtimeChannelImpl.kt b/Realtime/src/commonMain/kotlin/io/github/jan/supabase/realtime/RealtimeChannelImpl.kt
index 71197593b..3d11ef807 100644
--- a/Realtime/src/commonMain/kotlin/io/github/jan/supabase/realtime/RealtimeChannelImpl.kt
+++ b/Realtime/src/commonMain/kotlin/io/github/jan/supabase/realtime/RealtimeChannelImpl.kt
@@ -27,23 +27,23 @@ import kotlin.reflect.KClass
 import kotlin.reflect.KType
 
 internal class RealtimeChannelImpl(
-    private val realtimeImpl: RealtimeImpl,
+    override val realtime: Realtime,
     override val topic: String,
     private val broadcastJoinConfig: BroadcastJoinConfig,
     private val presenceJoinConfig: PresenceJoinConfig,
     private val isPrivate: Boolean,
 ) : RealtimeChannel {
 
+    private val realtimeImpl: RealtimeImpl = realtime as RealtimeImpl
     private val clientChanges = AtomicMutableList<PostgresJoinConfig>()
     @SupabaseInternal
     override val callbackManager = CallbackManagerImpl(realtimeImpl.serializer)
     private val _status = MutableStateFlow(RealtimeChannel.Status.UNSUBSCRIBED)
     override val status = _status.asStateFlow()
-    override val realtime: Realtime = realtimeImpl
     override val supabaseClient = realtimeImpl.supabaseClient
 
     private val broadcastUrl = realtimeImpl.broadcastUrl()
-    private val subTopic = topic.replaceFirst(Regex("^realtime:", RegexOption.IGNORE_CASE), "")
+    private val subTopic = topic.replaceFirst(Regex("^${RealtimeTopic.PREFIX}:", RegexOption.IGNORE_CASE), "")
     private val httpClient = realtimeImpl.supabaseClient.httpClient
 
     private suspend fun accessToken() = realtimeImpl.config.accessToken(supabaseClient) ?: realtimeImpl.accessToken
@@ -54,9 +54,6 @@ internal class RealtimeChannelImpl(
             if(!realtimeImpl.config.connectOnSubscribe) error("You can't subscribe to a channel while the realtime client is not connected. Did you forget to call `realtime.connect()`?")
             realtimeImpl.connect()
         }
-        realtimeImpl.run {
-            addChannel(this@RealtimeChannelImpl)
-        }
         _status.value = RealtimeChannel.Status.SUBSCRIBING
         Realtime.logger.d { "Subscribing to channel $topic" }
         val currentJwt = accessToken()
diff --git a/Realtime/src/commonMain/kotlin/io/github/jan/supabase/realtime/RealtimeImpl.kt b/Realtime/src/commonMain/kotlin/io/github/jan/supabase/realtime/RealtimeImpl.kt
index 28994f4bd..b36709574 100644
--- a/Realtime/src/commonMain/kotlin/io/github/jan/supabase/realtime/RealtimeImpl.kt
+++ b/Realtime/src/commonMain/kotlin/io/github/jan/supabase/realtime/RealtimeImpl.kt
@@ -1,7 +1,6 @@
 package io.github.jan.supabase.realtime
 
 import io.github.jan.supabase.SupabaseClient
-import io.github.jan.supabase.annotations.SupabaseInternal
 import io.github.jan.supabase.auth.Auth
 import io.github.jan.supabase.auth.status.SessionStatus
 import io.github.jan.supabase.buildUrl
@@ -162,6 +161,14 @@ import kotlin.io.encoding.ExperimentalEncodingApi
         _status.value = Realtime.Status.DISCONNECTED
     }
 
+    override fun channel(channelId: String, builder: RealtimeChannelBuilder): RealtimeChannel {
+        val topic = RealtimeTopic.withChannelId(channelId)
+        if(subscriptions.containsKey(topic)) return subscriptions[topic]!!
+        val channel = builder.build(this)
+        _subscriptions[topic] = channel
+        return channel
+    }
+
     private suspend fun onMessage(message: RealtimeMessage) {
         Realtime.logger.d { "Received message $message" }
         val channel = subscriptions[message.topic] as? RealtimeChannelImpl
@@ -234,11 +241,6 @@ import kotlin.io.encoding.ExperimentalEncodingApi
         }
     }
 
-    @SupabaseInternal
-    override fun Realtime.addChannel(channel: RealtimeChannel) {
-        _subscriptions[channel.topic] = channel
-    }
-
     override suspend fun close() {
         disconnect()
     }
diff --git a/Realtime/src/commonMain/kotlin/io/github/jan/supabase/realtime/RealtimeTopic.kt b/Realtime/src/commonMain/kotlin/io/github/jan/supabase/realtime/RealtimeTopic.kt
new file mode 100644
index 000000000..97bc9847d
--- /dev/null
+++ b/Realtime/src/commonMain/kotlin/io/github/jan/supabase/realtime/RealtimeTopic.kt
@@ -0,0 +1,12 @@
+package io.github.jan.supabase.realtime
+
+@PublishedApi
+internal object RealtimeTopic {
+
+    const val PREFIX = "realtime"
+
+    fun withChannelId(channelId: String): String {
+        return "$PREFIX:$channelId"
+    }
+
+}
\ No newline at end of file


Test patch:
diff --git a/Realtime/src/commonTest/kotlin/RealtimeTest.kt b/Realtime/src/commonTest/kotlin/RealtimeTest.kt
index 5ff705e64..d6689f253 100644
--- a/Realtime/src/commonTest/kotlin/RealtimeTest.kt
+++ b/Realtime/src/commonTest/kotlin/RealtimeTest.kt
@@ -1,6 +1,7 @@
 import io.github.jan.supabase.realtime.Realtime
 import io.github.jan.supabase.realtime.RealtimeImpl
 import io.github.jan.supabase.realtime.RealtimeMessage
+import io.github.jan.supabase.realtime.channel
 import io.github.jan.supabase.realtime.realtime
 import io.ktor.util.encodeBase64
 import kotlinx.coroutines.test.runTest
@@ -31,6 +32,22 @@ class RealtimeTest {
         }
     }
 
+    @Test
+    fun testExistingChannelShouldBeReturned() {
+        runTest {
+            createTestClient(
+                wsHandler = { _, _ ->
+                    //Does not matter for this test
+                },
+                supabaseHandler = {
+                    val channel = it.realtime.channel("channelId")
+                    val channel2 = it.realtime.channel("channelId")
+                    assertEquals(channel, channel2)
+                }
+            )
+        }
+    }
+
     @Test
     fun testSendingRealtimeMessages() {
         val expectedMessage = RealtimeMessage(
diff --git a/Realtime/src/commonTest/kotlin/RealtimeTestUtils.kt b/Realtime/src/commonTest/kotlin/RealtimeTestUtils.kt
index 34edb30fb..2250877f6 100644
--- a/Realtime/src/commonTest/kotlin/RealtimeTestUtils.kt
+++ b/Realtime/src/commonTest/kotlin/RealtimeTestUtils.kt
@@ -5,6 +5,7 @@ import io.github.jan.supabase.realtime.Presence
 import io.github.jan.supabase.realtime.RealtimeChannel.Companion.CHANNEL_EVENT_PRESENCE_DIFF
 import io.github.jan.supabase.realtime.RealtimeChannel.Companion.CHANNEL_EVENT_SYSTEM
 import io.github.jan.supabase.realtime.RealtimeMessage
+import io.github.jan.supabase.realtime.RealtimeTopic
 import kotlinx.coroutines.channels.ReceiveChannel
 import kotlinx.coroutines.channels.SendChannel
 import kotlinx.coroutines.flow.Flow
@@ -29,18 +30,18 @@ suspend fun Auth.importAuthTokenValid(token: String) {
 
 suspend fun handleSubscribe(incoming: ReceiveChannel<RealtimeMessage>, outgoing: SendChannel<RealtimeMessage>, channelId: String) {
     incoming.receive()
-    outgoing.send(RealtimeMessage("realtime:$channelId", CHANNEL_EVENT_SYSTEM, buildJsonObject { put("status", "ok") }, ""))
+    outgoing.send(RealtimeMessage(RealtimeTopic.withChannelId(channelId), CHANNEL_EVENT_SYSTEM, buildJsonObject { put("status", "ok") }, ""))
 }
 
 suspend fun SendChannel<RealtimeMessage>.sendBroadcast(channelId: String, event: String, message: JsonObject) {
-    send(RealtimeMessage("realtime:$channelId", "broadcast", buildJsonObject {
+    send(RealtimeMessage(RealtimeTopic.withChannelId(channelId), "broadcast", buildJsonObject {
         put("event", event)
         put("payload", message)
     }, ""))
 }
 
 suspend fun SendChannel<RealtimeMessage>.sendPresence(channelId: String, joins: Map<String, Presence>, leaves: Map<String, Presence>) {
-    send(RealtimeMessage("realtime:$channelId", CHANNEL_EVENT_PRESENCE_DIFF, buildJsonObject {
+    send(RealtimeMessage(RealtimeTopic.withChannelId(channelId), CHANNEL_EVENT_PRESENCE_DIFF, buildJsonObject {
         put("joins", transformPresenceMap(joins))
         put("leaves", transformPresenceMap(leaves))
     }, ""))
diff --git a/Realtime/src/commonTest/kotlin/RealtimeTopicTest.kt b/Realtime/src/commonTest/kotlin/RealtimeTopicTest.kt
new file mode 100644
index 000000000..a66a0a24d
--- /dev/null
+++ b/Realtime/src/commonTest/kotlin/RealtimeTopicTest.kt
@@ -0,0 +1,19 @@
+import io.github.jan.supabase.realtime.RealtimeTopic
+import kotlin.test.Test
+import kotlin.test.assertEquals
+
+class RealtimeTopicTest {
+
+    @Test
+    fun testRealtimeTopic() {
+        val channelId = "channelId"
+        val topic = RealtimeTopic.withChannelId(channelId)
+        assertEquals("realtime:channelId", topic)
+    }
+
+    @Test
+    fun testRealtimePrefix() {
+        assertEquals("realtime", RealtimeTopic.PREFIX)
+    }
+
+}
\ No newline at end of file