[pull] develop from baserow:develop

backend/src/baserow/config/settings/test.py

@@ -8,6 +8,8 @@
 
 from baserow.config.settings.utils import str_to_bool
 
+_fake_redis_server = FakeServer()
+
 # Test environment configuration:
 # - TEST_ENV_FILE: Optional env file for test settings (default: none)
 # - DATABASE_*, REDIS_*, and MEDIA_ROOT vars can be passed via real env vars or TEST_ENV_FILE
@@ -94,7 +96,7 @@ def getenv_for_tests(key: str, default: str = "") -> str:
         "OPTIONS": {
             "CONNECTION_POOL_KWARGS": {
                 "connection_class": FakeConnection,
-                "server": FakeServer(),
+                "server": _fake_redis_server,
             }
         },
         "KEY_PREFIX": "baserow-default-cache",

backend/src/baserow/contrib/database/apps.py

@@ -575,7 +575,11 @@ def ready(self):
 
         application_type_registry.register(DatabaseApplicationType())
 
-        from .ws.pages import PublicViewPageType, RowPageType, TablePageType
+        from .ws.pages import (
+            PublicViewPageType,
+            RowPageType,
+            TablePageType,
+        )
 
         page_registry.register(TablePageType())
         page_registry.register(PublicViewPageType())

backend/src/baserow/contrib/database/ws/pages.py

@@ -51,6 +51,22 @@ def get_group_name(self, table_id, **kwargs):
     def get_permission_channel_group_name(self, table_id, **kwargs):
         return f"permissions-table-{table_id}"
 
+    def get_presence_space_name(self, table_id: int | None, **kwargs) -> str | None:
+        return table_presence_space_name(table_id)
+
+
+def table_presence_space_name(table_id: int) -> str | None:
+    """
+    Return the canonical presence space name for a table.
+
+    :param table_id: The database table id.
+    :return: The space name string, or None if table_id is falsy.
+    """
+
+    if table_id is None or table_id <= 0:
+        return None
+    return f"table-{table_id}"
+
 
 class PublicViewPageType(PageType):
     type = "view"

backend/src/baserow/contrib/integrations/core/service_types.py

@@ -2075,7 +2075,7 @@ def _read_csv(
             return []
 
         if first_line_is_header:
-            headers = [header.strip() for header in csv_rows[0]]
+            headers = [self._normalize_csv_header(header) for header in csv_rows[0]]
             return [
                 self._row_to_dict(headers, row)
                 for row in csv_rows[1:]
@@ -2095,6 +2095,9 @@ def _row_to_dict(self, headers: list[str], row: list[str]) -> dict[str, str]:
             for index, header in enumerate(headers)
         }
 
+    def _normalize_csv_header(self, header: str) -> str:
+        return header.removeprefix("\ufeff").strip()
+
     def _row_has_values(self, row: list[str]) -> bool:
         return any(value != "" for value in row)
 

backend/src/baserow/core/async_redis.py

@@ -0,0 +1,40 @@
+import asyncio
+from weakref import WeakKeyDictionary
+
+from django.conf import settings
+
+from redis.asyncio import Redis, from_url
+
+_pools: WeakKeyDictionary[asyncio.AbstractEventLoop, Redis] = WeakKeyDictionary()
+_test_override: Redis | None = None
+
+
+async def get_async_redis() -> Redis:
+    """
+    Return a shared async Redis client for the current event loop.
+
+    Each loop gets its own client because ``redis.asyncio`` pins
+    connections to the loop that created them.  Dead loops are evicted
+    automatically via ``WeakKeyDictionary``.
+    """
+
+    if _test_override is not None:
+        return _test_override
+
+    loop = asyncio.get_running_loop()
+    client = _pools.get(loop)
+    if client is None:
+        client = from_url(settings.REDIS_URL, decode_responses=True)
+        _pools[loop] = client
+    return client
+
+
+def set_async_redis(client: Redis | None) -> None:
+    """
+    Replace the shared pool — used by tests to inject a fake client.
+
+    The override is loop-agnostic: it bypasses the per-loop map entirely.
+    """
+
+    global _test_override
+    _test_override = client

backend/src/baserow/core/feature_flags.py

@@ -3,6 +3,7 @@
 from baserow.core.exceptions import FeatureDisabledException
 
 FF_ENABLE_ALL = "*"
+FF_USER_PRESENCE = "user_presence"
 
 
 def feature_flag_is_enabled(feature_flag: str, raise_if_disabled=False) -> bool:

backend/src/baserow/ws/consumers.py

@@ -8,6 +8,12 @@
 from opentelemetry import metrics
 
 from baserow.config.settings.utils import try_int
+from baserow.core.feature_flags import FF_USER_PRESENCE, feature_flag_is_enabled
+from baserow.ws.presence import (
+    NullPresenceHandler,
+    PresenceHandler,
+    PresenceHandlerProtocol,
+)
 from baserow.ws.realtime_events import (
     FIRST_CONNECT_CURSOR,
     NO_REPLAY_AVAILABLE,
@@ -166,6 +172,8 @@ def __iter__(self):
 
 
 class CoreConsumer(AsyncJsonWebsocketConsumer):
+    presence: PresenceHandlerProtocol = NullPresenceHandler()
+
     async def connect(self):
         await self.accept()
         websocket_connections_counter.add(1)
@@ -188,6 +196,15 @@ async def connect(self):
             return
 
         self.scope["pages"] = SubscribedPages()
+        web_socket_id = self.scope["web_socket_id"]
+        if feature_flag_is_enabled(FF_USER_PRESENCE):
+            self.presence = PresenceHandler(
+                consumer=self,
+                web_socket_id=web_socket_id,
+                user_id=user.id,
+            )
+        else:
+            self.presence = NullPresenceHandler()
         await self.channel_layer.group_add("users", self.channel_name)
 
     async def disconnect(self, code):
@@ -196,7 +213,8 @@ async def disconnect(self, code):
         # eager connect counter would drift ahead during infrastructure failures,
         # exactly when the metric matters most.
         try:
-            await self._remove_all_page_scopes(send_confirmation=False)
+            await self.presence.leave_all_spaces()
+            await self._discard_all_channel_groups()
             await self.channel_layer.group_discard("users", self.channel_name)
         finally:
             websocket_disconnects_counter.add(
@@ -295,9 +313,15 @@ async def _add_page_scope(self, content: dict):
         self.scope["pages"].add(page_scope)
 
         await self.send_json(
-            {"type": "page_add", "page": page_type.type, "parameters": parameters}
+            {
+                "type": "page_add",
+                "page": page_type.type,
+                "parameters": parameters,
+            }
         )
 
+        await self.presence.handle_page_subscribed(page_type.type, parameters)
+
     async def _remove_page_scope(self, content: dict, send_confirmation=True):
         """
         Unsubscribes the connection from a page. Based on the provided page
@@ -315,17 +339,18 @@ async def _remove_page_scope(self, content: dict, send_confirmation=True):
         if not context:
             return
 
-        user, page_type, parameters = attrgetter(
-            "user", "resolved_page_type", "page_scope.page_parameters"
+        page_type, parameters = attrgetter(
+            "resolved_page_type", "page_scope.page_parameters"
         )(context)
 
         group_name = page_type.get_group_name(**parameters)
         await self.channel_layer.group_discard(group_name, self.channel_name)
 
         page_scope = PageScope(page_type=page_type.type, page_parameters=parameters)
-
         self.scope["pages"].remove(page_scope)
 
+        await self.presence.handle_page_unsubscribed(page_type.type, parameters)
+
         permission_group_name = page_type.get_permission_channel_group_name(
             **parameters
         )
@@ -348,18 +373,53 @@ async def _remove_page_scope(self, content: dict, send_confirmation=True):
     async def _remove_all_page_scopes(self, send_confirmation=True):
         """
         Unsubscribes the connection from all currently subscribed pages.
+        Used for live connections (e.g. permission revocation that affects
+        all pages). For disconnect cleanup, use _discard_all_channel_groups.
         """
 
         if self.scope.get("pages"):
             for page_scope in self.scope["pages"].copy():
-                content = {
-                    "user": self.scope["user"],
-                    "web_socket_id": self.scope["web_socket_id"],
-                    "remove_page": page_scope.page_type,
-                    **page_scope.page_parameters,
-                }
-                await self._remove_page_scope(
-                    content, send_confirmation=send_confirmation
+                try:
+                    content = {
+                        "user": self.scope["user"],
+                        "web_socket_id": self.scope["web_socket_id"],
+                        "remove_page": page_scope.page_type,
+                        **page_scope.page_parameters,
+                    }
+                    await self._remove_page_scope(
+                        content, send_confirmation=send_confirmation
+                    )
+                except Exception:
+                    logger.exception(
+                        "Failed to remove page scope {} during cleanup",
+                        page_scope.page_type,
+                    )
+
+    async def _discard_all_channel_groups(self) -> None:
+        """
+        Tears down all channel group memberships without sending client
+        messages or touching presence. Used during disconnect — presence
+        cleanup is handled separately by leave_all_spaces().
+        """
+
+        if not self.scope.get("pages"):
+            return
+
+        for page_scope in self.scope["pages"].copy():
+            try:
+                page_type = page_registry.get(page_scope.page_type)
+                group_name = page_type.get_group_name(**page_scope.page_parameters)
+                await self.channel_layer.group_discard(group_name, self.channel_name)
+                perm = page_type.get_permission_channel_group_name(
+                    **page_scope.page_parameters
+                )
+                if perm:
+                    await self.channel_layer.group_discard(perm, self.channel_name)
+            except Exception:
+                logger.exception(
+                    "Failed to discard channel groups for page scope {} "
+                    "during disconnect",
+                    page_scope.page_type,
                 )
 
     async def _remove_page_scopes_associated_with_perm_group(