Moved stream subscription lifecycle handling into SQLite.

owncastsentry/README.md

@@ -110,13 +110,15 @@ scheduled polling, and tests wired through explicit dependencies.
 handlers pass it user-supplied stream targets, and it turns those targets into
 normalized domains before creating or removing stored data.
 
-Subscribing a room creates a subscription for the normalized domain and creates
+Subscribing a room validates domains with no current subscribers through
-the shared stream record when needed. Domains with no current subscribers are
+`OwncastClient`, then inserts a subscription for the normalized domain. SQLite
-validated through `OwncastClient`; domains that already have subscribers reuse
+creates the shared stream record from that subscription insert when needed.
-the existing stream record instead of revalidating.
+Domains that already have subscribers reuse the existing stream record instead
+of revalidating.
 
-Unsubscribing removes one room's subscription to a domain. It does not delete
+Unsubscribing removes one room's subscription to a domain. When the last
-the shared stream record or make remote Owncast requests.
+subscription for a domain is removed, SQLite deletes the shared stream record.
+Unsubscribe does not make remote Owncast requests.
 
 Listing methods return the subscriptions for one Matrix room with shared stream
 state attached, so commands can display each instance's name, title, link,
@@ -287,15 +289,18 @@ Maubot owns the database connection and runs the schema upgrades registered by
 package's stream and subscription persistence operations.
 
 The persistence model stores one stream record per normalized domain and one
-subscription row per room/domain pair. `SubscriptionManager` normalizes user
+subscription row per room/domain pair. SQLite constraints enforce non-empty
-input before repository calls. `StreamRepository` writes display and state
+identifiers, unique room/domain subscriptions, and subscription ownership by a
-fields, while failure counters use dedicated methods.
+stream record. `SubscriptionManager` normalizes user input before repository
+calls. `StreamRepository` writes display and state fields, while failure
+counters use dedicated methods.
 `SubscriptionRepository` raises domain-specific errors for duplicate adds and
 missing removes.
 
-Room subscription listings join `subscriptions` to `streams`, which means
+Subscription inserts create missing stream rows through a SQLite trigger, and
-orphaned subscription entries without a matching stream record are skipped in
+deleting the last subscription for a domain deletes its stream row. Deleting a
-room display queries.
+stream cascades to its subscriptions through SQLite, so cleanup only has to
+remove the stream record after user notifications are sent.
 
 ## Metrics
 

owncastsentry/__init__.py

@@ -102,7 +102,6 @@ class OwncastSentry(Plugin):
         # Initialize subscription manager
         self.subscription_manager = SubscriptionManager(
             self.owncast_client,
-            self.stream_repo,
             self.subscription_repo,
             self.log,
         )

owncastsentry/repository.py

@@ -192,13 +192,134 @@ async def upgrade_v4(conn: Connection) -> None:
     await conn.execute("ALTER TABLE streams_new RENAME TO streams")
 
 
+@upgrade_table.register(  # type: ignore[arg-type, call-arg, untyped-decorator]
+    description="Formalize stream and subscription constraints"
+)
+async def upgrade_v5(conn: Connection) -> None:
+    """Upgrade database schema to version 5 format.
+
+    Rebuilds the current tables with explicit SQLite constraints, keeps only
+    stream rows that have valid subscriptions, and creates placeholder stream
+    rows for legacy subscriptions before adding the subscription-to-stream
+    foreign key.
+
+    Subscription inserts create their stream parent row in a trigger before the
+    foreign key is checked. Stream deletions cascade to dependent subscriptions.
+    Subscription deletions also run a trigger that removes the stream row once
+    no subscriptions remain.
+
+    :param conn: A connection to run the v5 database migration on.
+    """
+    await conn.execute(
+        """CREATE TABLE "streams_new" (
+            "domain" TEXT NOT NULL CHECK(length(trim("domain")) > 0),
+            "name" TEXT,
+            "title" TEXT,
+            "online" INTEGER NOT NULL DEFAULT 0 CHECK("online" IN (0, 1)),
+            "status_since" TEXT,
+            "failure_counter" INTEGER NOT NULL DEFAULT 0
+                CHECK("failure_counter" >= 0),
+            PRIMARY KEY("domain")
+        )"""
+    )
+
+    await conn.execute(
+        """INSERT INTO streams_new (
+               domain, name, title, online, status_since, failure_counter
+           )
+           SELECT domain,
+                  name,
+                  title,
+                  CASE WHEN online THEN 1 ELSE 0 END,
+                  status_since,
+                  CASE
+                      WHEN failure_counter IS NULL OR failure_counter < 0 THEN 0
+                      ELSE failure_counter
+                  END
+           FROM streams
+           WHERE domain IS NOT NULL
+             AND length(trim(domain)) > 0
+             AND EXISTS (
+                 SELECT 1
+                 FROM subscriptions
+                 WHERE subscriptions.stream_domain = streams.domain
+                   AND subscriptions.room_id IS NOT NULL
+                   AND length(trim(subscriptions.room_id)) > 0
+             )"""
+    )
+
+    await conn.execute(
+        """INSERT OR IGNORE INTO streams_new (domain)
+           SELECT DISTINCT stream_domain
+           FROM subscriptions
+           WHERE stream_domain IS NOT NULL
+             AND length(trim(stream_domain)) > 0
+             AND room_id IS NOT NULL
+             AND length(trim(room_id)) > 0"""
+    )
+
+    await conn.execute("DROP TABLE streams")
+    await conn.execute("ALTER TABLE streams_new RENAME TO streams")
+
+    await conn.execute(
+        """CREATE TABLE "subscriptions_new" (
+            "stream_domain" TEXT NOT NULL
+                CHECK(length(trim("stream_domain")) > 0),
+            "room_id" TEXT NOT NULL CHECK(length(trim("room_id")) > 0),
+            PRIMARY KEY("room_id", "stream_domain"),
+            FOREIGN KEY("stream_domain")
+                REFERENCES "streams"("domain")
+                ON DELETE CASCADE
+        )"""
+    )
+
+    await conn.execute(
+        """INSERT INTO subscriptions_new (stream_domain, room_id)
+           SELECT stream_domain, room_id
+           FROM subscriptions
+           WHERE stream_domain IS NOT NULL
+             AND length(trim(stream_domain)) > 0
+             AND room_id IS NOT NULL
+             AND length(trim(room_id)) > 0"""
+    )
+
+    await conn.execute("DROP TABLE subscriptions")
+    await conn.execute("ALTER TABLE subscriptions_new RENAME TO subscriptions")
+    await conn.execute(
+        """CREATE INDEX "subscriptions_stream_domain_idx"
+           ON "subscriptions"("stream_domain")"""
+    )
+    await conn.execute(
+        """CREATE TRIGGER "create_stream_for_subscription"
+           BEFORE INSERT ON "subscriptions"
+           BEGIN
+               INSERT INTO "streams" ("domain")
+               VALUES (NEW."stream_domain")
+               ON CONFLICT("domain") DO NOTHING;
+           END"""
+    )
+    await conn.execute(
+        """CREATE TRIGGER "delete_unsubscribed_stream"
+           AFTER DELETE ON "subscriptions"
+           WHEN NOT EXISTS (
+               SELECT 1
+               FROM "subscriptions"
+               WHERE "stream_domain" = OLD."stream_domain"
+           )
+           BEGIN
+               DELETE FROM "streams"
+               WHERE "domain" = OLD."stream_domain";
+           END"""
+    )
+
+
 def get_upgrade_table() -> UpgradeTable:
     """Return the repository upgrade table with registered migrations."""
     return upgrade_table
 
 
 class StreamRepository:
-    """Repository for managing stream data in the database."""
+    """Repository for stream parent rows and persisted stream state."""
 
     def __init__(self, database: Database) -> None:
         """Initialize the stream repository.
@@ -210,6 +331,10 @@ class StreamRepository:
     async def create(self, domain: str) -> bool:
         """Create a new stream entry in the database.
 
+        Subscription inserts also create stream rows through a SQLite trigger;
+        this explicit create method is for workflows that need a stream row
+        before any subscription is inserted.
+
         :param domain: The stream domain.
         :return: True if created, False if the stream already existed.
         """
@@ -231,15 +356,6 @@ class StreamRepository:
             row = await conn.fetchrow(query, domain)
             return StreamState.from_db_row(row) if row else None
 
-    async def exists(self, domain: str) -> bool:
-        """Check if a stream exists in the database.
-
-        :param domain: The stream domain.
-        :return: True if exists, False otherwise.
-        """
-        result = await self.get_by_domain(domain)
-        return result is not None
-
     async def update(
         self,
         domain: str,
@@ -302,6 +418,8 @@ class StreamRepository:
     async def delete(self, domain: str) -> None:
         """Delete a stream record from the database.
 
+        SQLite cascades the delete to all subscriptions for the stream.
+
         :param domain: The stream domain.
         """
         query = "DELETE FROM streams WHERE domain=$1"
@@ -332,7 +450,7 @@ class StreamRepository:
 
 
 class SubscriptionRepository:
-    """Repository for managing stream subscriptions in the database."""
+    """Repository for subscriptions and their database-owned stream rows."""
 
     def __init__(self, database: Database) -> None:
         """Initialize the subscription repository.
@@ -344,6 +462,10 @@ class SubscriptionRepository:
     async def add(self, domain: str, room_id: str) -> None:
         """Add a subscription for a room to a stream.
 
+        SQLite creates the stream parent row in a trigger before enforcing the
+        subscription foreign key. Callers must validate first subscriptions
+        before inserting untrusted domains.
+
         :param domain: The stream domain.
         :param room_id: The Matrix room ID.
         :raises AlreadySubscribedError: If subscription already exists.
@@ -359,6 +481,9 @@ class SubscriptionRepository:
     async def remove(self, domain: str, room_id: str) -> None:
         """Remove a subscription for a room from a stream.
 
+        SQLite removes the stream row through a trigger when this deletes the
+        last subscription for the domain.
+
         :param domain: The stream domain.
         :param room_id: The Matrix room ID.
         :raises NotSubscribedError: If no subscription exists.
@@ -369,17 +494,6 @@ class SubscriptionRepository:
             if int(result.rowcount) == 0:
                 raise NotSubscribedError(domain)
 
-    async def delete_all_for_domain(self, domain: str) -> int:
-        """Delete all subscriptions for a given stream domain.
-
-        :param domain: The stream domain.
-        :return: Number of subscriptions deleted.
-        """
-        query = "DELETE FROM subscriptions WHERE stream_domain=$1"
-        async with self.db.acquire() as conn:
-            result = await conn.execute(query, domain)
-            return int(result.rowcount)
-
     async def get_subscribed_rooms(self, domain: str) -> list[str]:
         """Get all room IDs subscribed to a stream.
 
@@ -391,16 +505,12 @@ class SubscriptionRepository:
             results = await conn.fetch(query, domain)
             return [row["room_id"] for row in results]
 
-    async def get_subscribed_streams_for_room(self, room_id: str) -> list[str]:
+    async def has_domain_subscriptions(self, domain: str) -> bool:
-        """Get all stream domains that a room is subscribed to.
+        """Check whether a stream domain has any subscriptions."""
-
+        query = "SELECT 1 FROM subscriptions WHERE stream_domain=$1 LIMIT 1"
-        :param room_id: The Matrix room ID.
-        :return: List of stream domains.
-        """
-        query = "SELECT stream_domain FROM subscriptions WHERE room_id=$1"
         async with self.db.acquire() as conn:
-            results = await conn.fetch(query, room_id)
+            result = await conn.fetchrow(query, domain)
-            return [row["stream_domain"] for row in results]
+            return result is not None
 
     async def has_room_subscriptions(self, room_id: str) -> bool:
         """Check whether a room has any subscriptions."""
@@ -419,7 +529,7 @@ class SubscriptionRepository:
                    FROM subscriptions
                    JOIN streams ON streams.domain = subscriptions.stream_domain
                    WHERE subscriptions.room_id=$1
-                   ORDER BY streams.domain"""
+                   ORDER BY subscriptions.stream_domain"""
         async with self.db.acquire() as conn:
             results = await conn.fetch(query, room_id)
             return [
@@ -438,7 +548,7 @@ class SubscriptionRepository:
                    WHERE subscriptions.room_id=$1
                      AND streams.online=true
                      AND streams.failure_counter <= $2
-                   ORDER BY streams.domain"""
+                   ORDER BY subscriptions.stream_domain"""
         async with self.db.acquire() as conn:
             results = await conn.fetch(query, room_id, UNKNOWN_STATUS_THRESHOLD)
             return [
@@ -459,17 +569,6 @@ class SubscriptionRepository:
             results = await conn.fetch(query)
             return [row["stream_domain"] for row in results]
 
-    async def count_by_domain(self, domain: str) -> int:
-        """Count the number of subscriptions for a given stream domain.
-
-        :param domain: The stream domain.
-        :return: Number of subscriptions.
-        """
-        query = "SELECT COUNT(*) FROM subscriptions WHERE stream_domain=$1"
-        async with self.db.acquire() as conn:
-            result = await conn.fetchrow(query, domain)
-            return int(result[0])
-
     async def count_by_domains(self, domains: list[str]) -> dict[str, int]:
         """Count subscriptions for each requested stream domain.
 
@@ -480,11 +579,16 @@ class SubscriptionRepository:
             return {}
 
         counts = dict.fromkeys(domains, 0)
+        placeholders = ", ".join(f"${index}" for index in range(1, len(counts) + 1))
         query = """SELECT stream_domain, COUNT(*) AS subscription_count
                    FROM subscriptions
+                   WHERE stream_domain IN ({placeholders})
                    GROUP BY stream_domain"""
         async with self.db.acquire() as conn:
-            results = await conn.fetch(query)
+            results = await conn.fetch(
+                query.format(placeholders=placeholders),
+                *counts,
+            )
 
         for row in results:
             domain = row["stream_domain"]

owncastsentry/stream_monitor.py

@@ -427,17 +427,13 @@ class StreamMonitor:
             # Send deletion notification
             await self.notification_service.send_cleanup_deletion(domain)
 
-            # Delete all subscriptions for this domain
-            deleted_count = await self.subscription_repo.delete_all_for_domain(domain)
-
             # Delete the stream record
             await self.stream_repo.delete(domain)
             self.offline_timer_cache.pop(domain, None)
             self.notification_service.clear_notification_state(domain)
 
             self.log.info(
-                "[%s] Cleanup complete. Deleted %s subscriptions and stream record.",
+                "[%s] Cleanup complete. Deleted stream record and subscriptions.",
                 domain,
-                deleted_count,
             )
             self.metrics.remove_stream(domain)

owncastsentry/subscription_manager.py

@@ -27,7 +27,7 @@ if TYPE_CHECKING:
     import logging
 
     from .owncast_client import OwncastClient
-    from .repository import StreamRepository, SubscriptionRepository
+    from .repository import SubscriptionRepository
 
 
 _DOMAIN_CLEANUP_RE = re.compile(r"[^a-z0-9.-]")
@@ -55,19 +55,20 @@ class SubscriptionManager:
     def __init__(
         self,
         owncast_client: OwncastClient,
-        stream_repo: StreamRepository,
         subscription_repo: SubscriptionRepository,
         logger: logging.Logger,
     ) -> None:
         """Initialize the subscription manager."""
         self.owncast_client = owncast_client
-        self.stream_repo = stream_repo
         self.subscription_repo = subscription_repo
         self.log = logger
 
     async def subscribe(self, room_id: str, url: str) -> str:
         """Subscribe a room to stream notifications and return the stream domain.
 
+        First subscriptions validate the remote Owncast instance before the
+        subscription insert creates the shared stream row through SQLite.
+
         :param room_id: Matrix room ID to subscribe.
         :param url: User-supplied Owncast URL, domain, or Fediverse-style address.
         :return: Normalized stream domain.
@@ -76,23 +77,27 @@ class SubscriptionManager:
         """
         stream_domain = _domainify(url)
 
-        subscription_count = await self.subscription_repo.count_by_domain(stream_domain)
+        is_new_domain = not await self.subscription_repo.has_domain_subscriptions(
-        if subscription_count == 0:
+            stream_domain
+        )
+        if is_new_domain:
             is_valid = await self.owncast_client.validate_instance(stream_domain)
             if not is_valid:
                 raise InvalidOwncastInstanceError(stream_domain)
 
         await self.subscription_repo.add(stream_domain, room_id)
 
-        if await self.stream_repo.create(stream_domain):
+        if is_new_domain:
             self.log.info("[%s] Discovered new stream!", stream_domain)
-
         self.log.info("[%s] Subscription added for room %s.", stream_domain, room_id)
         return stream_domain
 
     async def unsubscribe(self, room_id: str, url: str) -> str:
         """Remove a room subscription and return the stream domain.
 
+        When this removes the last subscription for a stream, SQLite deletes the
+        stream row through the subscription cleanup trigger.
+
         :param room_id: Matrix room ID to unsubscribe.
         :param url: User-supplied Owncast URL, domain, or Fediverse-style address.
         :return: Normalized stream domain.

tests/test_commands.py

@@ -40,6 +40,62 @@ if TYPE_CHECKING:
     from owncastsentry import OwncastSentry
 
 
+_VALID_STATUS_BODY = json.dumps(VALID_STATUS_RESPONSE).encode()
+
+
+async def _subscribe_valid_stream(
+    maubot_test_bot: TestBot,
+    domain: str = "stream.logal.dev",
+    *,
+    room_id: str | None = None,
+) -> None:
+    """Subscribe to a domain with a mocked valid Owncast status response."""
+    with aioresponses() as mocked:
+        mocked.get(
+            f"https://{domain}{_OWNCAST_STATUS_PATH}",
+            body=_VALID_STATUS_BODY,
+        )
+        if room_id is None:
+            await maubot_test_bot.send(f"!subscribe {domain}")
+        else:
+            await maubot_test_bot.send(f"!subscribe {domain}", room_id=room_id)
+
+
+async def _subscribe_valid_streams(
+    maubot_test_bot: TestBot,
+    *domains: str,
+) -> None:
+    """Subscribe to multiple domains with mocked valid Owncast status responses."""
+    with aioresponses() as mocked:
+        for domain in domains:
+            mocked.get(
+                f"https://{domain}{_OWNCAST_STATUS_PATH}",
+                body=_VALID_STATUS_BODY,
+            )
+
+        for domain in domains:
+            await maubot_test_bot.send(f"!subscribe {domain}")
+
+
+async def _set_stream_state(
+    maubot_plugin: OwncastSentry,
+    domain: str = "stream.logal.dev",
+    *,
+    name: str | None = None,
+    title: str | None = None,
+    online: bool = False,
+    status_since: str | None = None,
+) -> None:
+    """Write stream state used by command rendering tests."""
+    await maubot_plugin.stream_repo.update(
+        domain,
+        name=name,
+        title=title,
+        online=online,
+        status_since=status_since,
+    )
+
+
 class TestEscapeMarkdown:
     """Markdown special character escaping."""
 
@@ -121,13 +177,7 @@ class TestSubscribeCommand:
 
     async def test_subscribe_valid_stream(self, maubot_test_bot: TestBot) -> None:
         """Subscribe to a valid Owncast stream."""
-        status_url = f"https://stream.logal.dev{_OWNCAST_STATUS_PATH}"
+        await _subscribe_valid_stream(maubot_test_bot)
-        with aioresponses() as mocked:
-            mocked.get(
-                status_url,
-                body=json.dumps(VALID_STATUS_RESPONSE).encode(),
-            )
-            await maubot_test_bot.send("!subscribe stream.logal.dev")
 
         assert len(maubot_test_bot.responded) == 1
         assert maubot_test_bot.responded[0].content.body == (
@@ -152,13 +202,7 @@ class TestSubscribeCommand:
 
     async def test_subscribe_already_subscribed(self, maubot_test_bot: TestBot) -> None:
         """Reject duplicate subscription in the same room."""
-        status_url = f"https://stream.logal.dev{_OWNCAST_STATUS_PATH}"
+        await _subscribe_valid_stream(maubot_test_bot)
-        with aioresponses() as mocked:
-            mocked.get(
-                status_url,
-                body=json.dumps(VALID_STATUS_RESPONSE).encode(),
-            )
-            await maubot_test_bot.send("!subscribe stream.logal.dev")
 
         # Second subscribe; stream already exists so validation is skipped
         await maubot_test_bot.send("!subscribe stream.logal.dev")
@@ -172,13 +216,7 @@ class TestSubscribeCommand:
         self, maubot_test_bot: TestBot
     ) -> None:
         """Skip validation when the domain already has subscriptions."""
-        status_url = f"https://stream.logal.dev{_OWNCAST_STATUS_PATH}"
+        await _subscribe_valid_stream(maubot_test_bot)
-        with aioresponses() as mocked:
-            mocked.get(
-                status_url,
-                body=json.dumps(VALID_STATUS_RESPONSE).encode(),
-            )
-            await maubot_test_bot.send("!subscribe stream.logal.dev")
 
         # Subscribe from a different room. The existing subscribed domain skips
         # validation, so an empty aioresponses context will raise ConnectionError
@@ -201,13 +239,7 @@ class TestUnsubscribeCommand:
 
     async def test_unsubscribe_existing(self, maubot_test_bot: TestBot) -> None:
         """Unsubscribe from a subscribed stream."""
-        status_url = f"https://stream.logal.dev{_OWNCAST_STATUS_PATH}"
+        await _subscribe_valid_stream(maubot_test_bot)
-        with aioresponses() as mocked:
-            mocked.get(
-                status_url,
-                body=json.dumps(VALID_STATUS_RESPONSE).encode(),
-            )
-            await maubot_test_bot.send("!subscribe stream.logal.dev")
 
         await maubot_test_bot.send("!unsubscribe stream.logal.dev")
 
@@ -247,18 +279,9 @@ class TestSubscriptionsCommand:
         self, maubot_test_bot: TestBot, maubot_plugin: OwncastSentry
     ) -> None:
         """Show stream details including title and duration."""
-        # Subscribe first
+        await _subscribe_valid_stream(maubot_test_bot)
-        status_url = f"https://stream.logal.dev{_OWNCAST_STATUS_PATH}"
+        await _set_stream_state(
-        with aioresponses() as mocked:
+            maubot_plugin,
-            mocked.get(
-                status_url,
-                body=json.dumps(VALID_STATUS_RESPONSE).encode(),
-            )
-            await maubot_test_bot.send("!subscribe stream.logal.dev")
-
-        # Update stream state to be online with a name
-        await maubot_plugin.stream_repo.update(
-            "stream.logal.dev",
             name="Test Stream",
             title="Playing Games",
             online=True,
@@ -283,16 +306,9 @@ class TestSubscriptionsCommand:
         self, maubot_test_bot: TestBot, maubot_plugin: OwncastSentry
     ) -> None:
         """Render stream name and title as literal text in command output."""
-        status_url = f"https://stream.logal.dev{_OWNCAST_STATUS_PATH}"
+        await _subscribe_valid_stream(maubot_test_bot)
-        with aioresponses() as mocked:
+        await _set_stream_state(
-            mocked.get(
+            maubot_plugin,
-                status_url,
-                body=json.dumps(VALID_STATUS_RESPONSE).encode(),
-            )
-            await maubot_test_bot.send("!subscribe stream.logal.dev")
-
-        await maubot_plugin.stream_repo.update(
-            "stream.logal.dev",
             name="*Bold* [link](https://evil.example)\nName",
             title="`code` > quote #tag",
             online=True,
@@ -317,18 +333,9 @@ class TestSubscriptionsCommand:
         self, maubot_test_bot: TestBot, maubot_plugin: OwncastSentry
     ) -> None:
         """Show offline status for non-live streams."""
-        # Subscribe first
+        await _subscribe_valid_stream(maubot_test_bot)
-        status_url = f"https://stream.logal.dev{_OWNCAST_STATUS_PATH}"
+        await _set_stream_state(
-        with aioresponses() as mocked:
+            maubot_plugin,
-            mocked.get(
-                status_url,
-                body=json.dumps(VALID_STATUS_RESPONSE).encode(),
-            )
-            await maubot_test_bot.send("!subscribe stream.logal.dev")
-
-        # Update stream state to be offline
-        await maubot_plugin.stream_repo.update(
-            "stream.logal.dev",
             name="Test Stream",
             status_since="2026-01-01T10:00:00+00:00",
         )
@@ -349,13 +356,7 @@ class TestSubscriptionsCommand:
         self, maubot_test_bot: TestBot
     ) -> None:
         """Show offline status without duration before first poll completes."""
-        status_url = f"https://stream.logal.dev{_OWNCAST_STATUS_PATH}"
+        await _subscribe_valid_stream(maubot_test_bot)
-        with aioresponses() as mocked:
-            mocked.get(
-                status_url,
-                body=json.dumps(VALID_STATUS_RESPONSE).encode(),
-            )
-            await maubot_test_bot.send("!subscribe stream.logal.dev")
 
         # Stream row exists with no state yet - query subscriptions immediately
         await maubot_test_bot.send("!subscriptions")
@@ -374,13 +375,7 @@ class TestSubscriptionsCommand:
         self, maubot_test_bot: TestBot, maubot_plugin: OwncastSentry
     ) -> None:
         """Show unknown status when instance has been unreachable."""
-        status_url = f"https://stream.logal.dev{_OWNCAST_STATUS_PATH}"
+        await _subscribe_valid_stream(maubot_test_bot)
-        with aioresponses() as mocked:
-            mocked.get(
-                status_url,
-                body=json.dumps(VALID_STATUS_RESPONSE).encode(),
-            )
-            await maubot_test_bot.send("!subscribe stream.logal.dev")
 
         # Increment failure counter past the unknown threshold
         for _ in range(UNKNOWN_STATUS_THRESHOLD + 1):
@@ -406,27 +401,18 @@ class TestSubscriptionsCommand:
     ) -> None:
         """List subscriptions ordered by domain with mixed statuses."""
         # Subscribe in reverse domain order to verify domain-sorted output
-        with aioresponses() as mocked:
+        await _subscribe_valid_streams(maubot_test_bot, "beta.com", "alpha.com")
-            mocked.get(
-                f"https://beta.com{_OWNCAST_STATUS_PATH}",
-                body=json.dumps(VALID_STATUS_RESPONSE).encode(),
-            )
-            mocked.get(
-                f"https://alpha.com{_OWNCAST_STATUS_PATH}",
-                body=json.dumps(VALID_STATUS_RESPONSE).encode(),
-            )
-            await maubot_test_bot.send("!subscribe beta.com")
-            await maubot_test_bot.send("!subscribe alpha.com")
 
-        # Set alpha online, beta offline
+        await _set_stream_state(
-        await maubot_plugin.stream_repo.update(
+            maubot_plugin,
             "alpha.com",
             name="Alpha Stream",
             title="Streaming Live",
             online=True,
             status_since="2026-03-13T10:00:00+00:00",
         )
-        await maubot_plugin.stream_repo.update(
+        await _set_stream_state(
+            maubot_plugin,
             "beta.com",
             name="Beta Stream",
             status_since="2026-03-12T18:00:00+00:00",
@@ -470,18 +456,9 @@ class TestLiveCommand:
         self, maubot_test_bot: TestBot, maubot_plugin: OwncastSentry
     ) -> None:
         """Show 'no live' message when all streams are offline."""
-        # Subscribe first
+        await _subscribe_valid_stream(maubot_test_bot)
-        status_url = f"https://stream.logal.dev{_OWNCAST_STATUS_PATH}"
+        await _set_stream_state(
-        with aioresponses() as mocked:
+            maubot_plugin,
-            mocked.get(
-                status_url,
-                body=json.dumps(VALID_STATUS_RESPONSE).encode(),
-            )
-            await maubot_test_bot.send("!subscribe stream.logal.dev")
-
-        # Update stream state to offline
-        await maubot_plugin.stream_repo.update(
-            "stream.logal.dev",
             name="Test Stream",
             status_since="2026-01-01T10:00:00+00:00",
         )
@@ -500,18 +477,9 @@ class TestLiveCommand:
         self, maubot_test_bot: TestBot, maubot_plugin: OwncastSentry
     ) -> None:
         """Show live stream with title and duration."""
-        # Subscribe first
+        await _subscribe_valid_stream(maubot_test_bot)
-        status_url = f"https://stream.logal.dev{_OWNCAST_STATUS_PATH}"
+        await _set_stream_state(
-        with aioresponses() as mocked:
+            maubot_plugin,
-            mocked.get(
-                status_url,
-                body=json.dumps(VALID_STATUS_RESPONSE).encode(),
-            )
-            await maubot_test_bot.send("!subscribe stream.logal.dev")
-
-        # Update stream state to online
-        await maubot_plugin.stream_repo.update(
-            "stream.logal.dev",
             name="Test Stream",
             title="Playing Games",
             online=True,
@@ -535,27 +503,18 @@ class TestLiveCommand:
     ) -> None:
         """List live streams ordered by domain with different durations."""
         # Subscribe in reverse domain order to verify domain-sorted output
-        with aioresponses() as mocked:
+        await _subscribe_valid_streams(maubot_test_bot, "beta.com", "alpha.com")
-            mocked.get(
-                f"https://beta.com{_OWNCAST_STATUS_PATH}",
-                body=json.dumps(VALID_STATUS_RESPONSE).encode(),
-            )
-            mocked.get(
-                f"https://alpha.com{_OWNCAST_STATUS_PATH}",
-                body=json.dumps(VALID_STATUS_RESPONSE).encode(),
-            )
-            await maubot_test_bot.send("!subscribe beta.com")
-            await maubot_test_bot.send("!subscribe alpha.com")
 
-        # Set both streams online with different status timestamps
+        await _set_stream_state(
-        await maubot_plugin.stream_repo.update(
+            maubot_plugin,
             "alpha.com",
             name="Alpha Stream",
             title="Morning Show",
             online=True,
             status_since="2026-03-13T10:00:00+00:00",
         )
-        await maubot_plugin.stream_repo.update(
+        await _set_stream_state(
+            maubot_plugin,
             "beta.com",
             name="Beta Stream",
             title="Evening Vibes",

tests/test_repository.py

@@ -14,11 +14,15 @@
 
 """Tests for database repository classes."""
 
+import sqlite3
+from contextlib import closing
+from datetime import UTC, datetime
 from typing import TYPE_CHECKING
 
 import pytest
+from mautrix.util.async_db import Database
 
-from owncastsentry.repository import _normalize_legacy_status_since
+from owncastsentry.repository import _normalize_legacy_status_since, get_upgrade_table
 from owncastsentry.types import (
     UNKNOWN_STATUS_THRESHOLD,
     AlreadySubscribedError,
@@ -26,7 +30,7 @@ from owncastsentry.types import (
 )
 
 if TYPE_CHECKING:
-    from mautrix.util.async_db import Database
+    from pathlib import Path
 
     from owncastsentry.repository import StreamRepository, SubscriptionRepository
 
@@ -50,6 +54,576 @@ class TestStreamSchema:
             "failure_counter",
         ]
 
+    async def test_stream_constraint_metadata(self, database: Database) -> None:
+        """Declare stream primary key and non-null constrained columns."""
+        async with database.acquire() as conn:
+            rows = await conn.fetch("PRAGMA table_info(streams)")
+            columns = {row["name"]: row for row in rows}
+
+            assert columns["domain"]["pk"] == 1
+            assert columns["domain"]["notnull"] == 1
+            assert columns["online"]["notnull"] == 1
+            assert columns["failure_counter"]["notnull"] == 1
+
+    @pytest.mark.parametrize(
+        ("query", "params"),
+        [
+            pytest.param(
+                "INSERT INTO streams (domain) VALUES ($1)",
+                (None,),
+                id="null-domain",
+            ),
+            pytest.param(
+                "INSERT INTO streams (domain) VALUES ($1)",
+                ("",),
+                id="blank-domain",
+            ),
+            pytest.param(
+                "INSERT INTO streams (domain) VALUES ($1)",
+                ("   ",),
+                id="whitespace-domain",
+            ),
+            pytest.param(
+                "INSERT INTO streams (domain, online) VALUES ($1, $2)",
+                ("bad-online.example", 2),
+                id="invalid-online",
+            ),
+            pytest.param(
+                """INSERT INTO streams (domain, failure_counter)
+                   VALUES ($1, $2)""",
+                ("bad-counter.example", -1),
+                id="negative-failure-counter",
+            ),
+        ],
+    )
+    async def test_rejects_invalid_stream_values(
+        self,
+        database: Database,
+        query: str,
+        params: tuple[object, ...],
+    ) -> None:
+        """Reject stream rows that violate current value constraints."""
+        async with database.acquire() as conn:
+            with pytest.raises(sqlite3.IntegrityError):
+                await conn.execute(query, *params)
+
+
+class TestSubscriptionSchema:
+    """Subscriptions table schema after migrations."""
+
+    async def test_subscription_constraint_metadata(self, database: Database) -> None:
+        """Declare subscription key, foreign key, index, and lifecycle triggers."""
+        async with database.acquire() as conn:
+            rows = await conn.fetch("PRAGMA table_info(subscriptions)")
+            columns = {row["name"]: row for row in rows}
+            foreign_keys = await conn.fetch("PRAGMA foreign_key_list(subscriptions)")
+            indexes = await conn.fetch("PRAGMA index_list(subscriptions)")
+            index_names = {row["name"] for row in indexes}
+            domain_index_columns = await conn.fetch(
+                "PRAGMA index_info(subscriptions_stream_domain_idx)"
+            )
+            triggers = await conn.fetch(
+                "SELECT name FROM sqlite_schema WHERE type='trigger'"
+            )
+            trigger_names = {row["name"] for row in triggers}
+
+            assert columns["room_id"]["pk"] == 1
+            assert columns["stream_domain"]["pk"] == 2
+            assert columns["room_id"]["notnull"] == 1
+            assert columns["stream_domain"]["notnull"] == 1
+            assert "subscriptions_stream_domain_idx" in index_names
+            assert [row["name"] for row in domain_index_columns] == ["stream_domain"]
+            assert len(foreign_keys) == 1
+            assert foreign_keys[0]["table"] == "streams"
+            assert foreign_keys[0]["from"] == "stream_domain"
+            assert foreign_keys[0]["to"] == "domain"
+            assert foreign_keys[0]["on_delete"] == "CASCADE"
+            assert "create_stream_for_subscription" in trigger_names
+            assert "delete_unsubscribed_stream" in trigger_names
+
+    @pytest.mark.parametrize(
+        ("stream_domain", "room_id"),
+        [
+            pytest.param(None, "!null-domain:example.com", id="null-domain"),
+            pytest.param("", "!blank-domain:example.com", id="blank-domain"),
+            pytest.param("   ", "!blank-domain:example.com", id="whitespace-domain"),
+            pytest.param("example.com", None, id="null-room"),
+            pytest.param("example.com", "", id="blank-room"),
+            pytest.param("example.com", "   ", id="whitespace-room"),
+        ],
+    )
+    async def test_rejects_invalid_subscription_values(
+        self,
+        database: Database,
+        stream_domain: str | None,
+        room_id: str | None,
+    ) -> None:
+        """Reject subscription rows that violate current constraints."""
+        async with database.acquire() as conn:
+            await conn.execute(
+                "INSERT INTO streams (domain) VALUES ($1)",
+                "example.com",
+            )
+
+            with pytest.raises(sqlite3.IntegrityError):
+                await conn.execute(
+                    """INSERT INTO subscriptions (stream_domain, room_id)
+                       VALUES ($1, $2)""",
+                    stream_domain,
+                    room_id,
+                )
+
+    async def test_subscription_insert_creates_stream_parent(
+        self, database: Database
+    ) -> None:
+        """Create the stream parent row from the subscription insert trigger."""
+        async with database.acquire() as conn:
+            await conn.execute(
+                """INSERT INTO subscriptions (stream_domain, room_id)
+                   VALUES ($1, $2)""",
+                "missing.example",
+                "!room:example.com",
+            )
+            stream = await conn.fetchrow(
+                "SELECT * FROM streams WHERE domain=$1",
+                "missing.example",
+            )
+
+        assert stream is not None
+        assert stream["online"] == 0
+        assert stream["failure_counter"] == 0
+
+    async def test_subscription_insert_rolls_back_created_stream_on_failure(
+        self, database: Database
+    ) -> None:
+        """Roll back the trigger-created stream when the subscription is invalid."""
+        async with database.acquire() as conn:
+            with pytest.raises(sqlite3.IntegrityError):
+                await conn.execute(
+                    """INSERT INTO subscriptions (stream_domain, room_id)
+                       VALUES ($1, $2)""",
+                    "rollback.example",
+                    "",
+                )
+
+            stream = await conn.fetchrow(
+                "SELECT * FROM streams WHERE domain=$1",
+                "rollback.example",
+            )
+            subscription = await conn.fetchrow(
+                """SELECT * FROM subscriptions
+                   WHERE stream_domain=$1""",
+                "rollback.example",
+            )
+
+        assert stream is None
+        assert subscription is None
+
+    async def test_subscription_primary_key_rejects_duplicates(
+        self, database: Database
+    ) -> None:
+        """Reject duplicate room/domain subscription rows."""
+        async with database.acquire() as conn:
+            await conn.execute(
+                "INSERT INTO streams (domain) VALUES ($1)",
+                "example.com",
+            )
+            await conn.execute(
+                """INSERT INTO subscriptions (stream_domain, room_id)
+                   VALUES ($1, $2)""",
+                "example.com",
+                "!room:example.com",
+            )
+
+            with pytest.raises(sqlite3.IntegrityError):
+                await conn.execute(
+                    """INSERT INTO subscriptions (stream_domain, room_id)
+                       VALUES ($1, $2)""",
+                    "example.com",
+                    "!room:example.com",
+                )
+
+
+class TestSubscriptionLifecycleConstraints:
+    """Database-owned cleanup behavior between streams and subscriptions."""
+
+    async def test_stream_delete_cascades_subscriptions(
+        self,
+        stream_repo: StreamRepository,
+        subscription_repo: SubscriptionRepository,
+    ) -> None:
+        """Delete dependent subscription rows when a stream is deleted."""
+        await stream_repo.create("example.com")
+        await subscription_repo.add("example.com", "!room:example.com")
+
+        await stream_repo.delete("example.com")
+
+        assert await subscription_repo.get_subscribed_rooms("example.com") == []
+
+    async def test_last_subscription_delete_removes_stream(
+        self,
+        stream_repo: StreamRepository,
+        subscription_repo: SubscriptionRepository,
+    ) -> None:
+        """Delete the stream row when its last subscription is removed."""
+        await stream_repo.create("example.com")
+        await subscription_repo.add("example.com", "!room:example.com")
+
+        await subscription_repo.remove("example.com", "!room:example.com")
+
+        assert await stream_repo.get_by_domain("example.com") is None
+
+    async def test_subscription_delete_keeps_stream_with_remaining_rooms(
+        self,
+        stream_repo: StreamRepository,
+        subscription_repo: SubscriptionRepository,
+    ) -> None:
+        """Keep stream state while at least one subscription remains."""
+        await stream_repo.create("example.com")
+        await subscription_repo.add("example.com", "!room1:example.com")
+        await subscription_repo.add("example.com", "!room2:example.com")
+
+        await subscription_repo.remove("example.com", "!room1:example.com")
+
+        assert await stream_repo.get_by_domain("example.com") is not None
+        assert await subscription_repo.get_subscribed_rooms("example.com") == [
+            "!room2:example.com"
+        ]
+
+
+class TestConstraintMigration:
+    """Migration behavior for legacy data that did not have constraints."""
+
+    async def test_converts_legacy_status_timestamps_from_v3(
+        self, tmp_path: Path
+    ) -> None:
+        """Convert legacy connect/disconnect timestamps during v4 migration."""
+        db_path = tmp_path / "legacy.db"
+        with closing(sqlite3.connect(db_path)) as conn, conn:
+            conn.executescript(
+                """
+                CREATE TABLE version (version INTEGER PRIMARY KEY);
+                INSERT INTO version (version) VALUES (3);
+
+                CREATE TABLE streams (
+                    domain TEXT NOT NULL UNIQUE,
+                    name TEXT,
+                    title TEXT,
+                    last_connect_time TEXT,
+                    last_disconnect_time TEXT,
+                    failure_counter INTEGER DEFAULT 0,
+                    PRIMARY KEY(domain)
+                );
+
+                CREATE TABLE subscriptions (
+                    stream_domain TEXT NOT NULL,
+                    room_id TEXT NOT NULL,
+                    UNIQUE(room_id, stream_domain)
+                );
+
+                INSERT INTO streams (
+                    domain,
+                    last_connect_time,
+                    last_disconnect_time,
+                    failure_counter
+                )
+                VALUES
+                    (
+                        'live.example',
+                        '2026-05-21T19:06:24Z',
+                        '2026-05-20T00:00:00+00:00',
+                        2
+                    ),
+                    (
+                        'offline.example',
+                        NULL,
+                        '2026-05-17T21:23:27-04:00',
+                        3
+                    ),
+                    (
+                        'blank-connect.example',
+                        '',
+                        '2026-05-20T12:34:56+02:00',
+                        4
+                    ),
+                    (
+                        'invalid-timestamp.example',
+                        NULL,
+                        'not a timestamp',
+                        5
+                    );
+
+                INSERT INTO subscriptions (stream_domain, room_id)
+                VALUES
+                    ('live.example', '!live:example.com'),
+                    ('offline.example', '!offline:example.com'),
+                    ('blank-connect.example', '!blank:example.com'),
+                    ('invalid-timestamp.example', '!invalid:example.com');
+                """
+            )
+
+        db = Database.create(f"sqlite:///{db_path}", upgrade_table=get_upgrade_table())
+        await db.start()
+        try:
+            async with db.acquire() as conn:
+                rows = await conn.fetch(
+                    """SELECT domain, online, status_since, failure_counter
+                       FROM streams
+                       ORDER BY domain"""
+                )
+        finally:
+            await db.stop()
+
+        streams = {row["domain"]: row for row in rows}
+        assert streams["live.example"]["online"] == 1
+        assert streams["live.example"]["status_since"] == "2026-05-21T19:06:24+00:00"
+        assert streams["live.example"]["failure_counter"] == 2
+        assert streams["offline.example"]["online"] == 0
+        assert streams["offline.example"]["status_since"] == "2026-05-18T01:23:27+00:00"
+        assert streams["offline.example"]["failure_counter"] == 3
+        assert streams["blank-connect.example"]["online"] == 0
+        assert (
+            streams["blank-connect.example"]["status_since"]
+            == "2026-05-20T10:34:56+00:00"
+        )
+        assert streams["invalid-timestamp.example"]["online"] == 0
+        assert streams["invalid-timestamp.example"]["status_since"] is None
+
+    async def test_preserves_legacy_orphaned_subscription(self, tmp_path: Path) -> None:
+        """Create placeholder streams before enforcing subscription foreign keys."""
+        db_path = tmp_path / "legacy.db"
+        with closing(sqlite3.connect(db_path)) as conn, conn:
+            conn.executescript(
+                """
+                CREATE TABLE version (version INTEGER PRIMARY KEY);
+                INSERT INTO version (version) VALUES (4);
+
+                CREATE TABLE streams (
+                    domain TEXT NOT NULL UNIQUE,
+                    name TEXT,
+                    title TEXT,
+                    online BOOLEAN NOT NULL DEFAULT false,
+                    status_since TEXT,
+                    failure_counter INTEGER DEFAULT 0,
+                    PRIMARY KEY(domain)
+                );
+
+                CREATE TABLE subscriptions (
+                    stream_domain TEXT NOT NULL,
+                    room_id TEXT NOT NULL,
+                    UNIQUE(room_id, stream_domain)
+                );
+
+                INSERT INTO subscriptions (stream_domain, room_id)
+                VALUES ('missing.example', '!room:example.com');
+                """
+            )
+
+        db = Database.create(f"sqlite:///{db_path}", upgrade_table=get_upgrade_table())
+        await db.start()
+        try:
+            async with db.acquire() as conn:
+                stream = await conn.fetchrow(
+                    "SELECT * FROM streams WHERE domain=$1",
+                    "missing.example",
+                )
+                subscription = await conn.fetchrow(
+                    """SELECT * FROM subscriptions
+                       WHERE stream_domain=$1 AND room_id=$2""",
+                    "missing.example",
+                    "!room:example.com",
+                )
+        finally:
+            await db.stop()
+
+        assert stream is not None
+        assert stream["online"] == 0
+        assert stream["failure_counter"] == 0
+        assert subscription is not None
+
+    async def test_removes_legacy_stream_without_subscriptions(
+        self, tmp_path: Path
+    ) -> None:
+        """Drop stream rows that are no longer referenced by subscriptions."""
+        db_path = tmp_path / "legacy.db"
+        with closing(sqlite3.connect(db_path)) as conn, conn:
+            conn.executescript(
+                """
+                CREATE TABLE version (version INTEGER PRIMARY KEY);
+                INSERT INTO version (version) VALUES (4);
+
+                CREATE TABLE streams (
+                    domain TEXT NOT NULL UNIQUE,
+                    name TEXT,
+                    title TEXT,
+                    online BOOLEAN NOT NULL DEFAULT false,
+                    status_since TEXT,
+                    failure_counter INTEGER DEFAULT 0,
+                    PRIMARY KEY(domain)
+                );
+
+                CREATE TABLE subscriptions (
+                    stream_domain TEXT NOT NULL,
+                    room_id TEXT NOT NULL,
+                    UNIQUE(room_id, stream_domain)
+                );
+
+                INSERT INTO streams (domain)
+                VALUES ('subscribed.example'), ('orphaned.example');
+                INSERT INTO subscriptions (stream_domain, room_id)
+                VALUES ('subscribed.example', '!room:example.com');
+                """
+            )
+
+        db = Database.create(f"sqlite:///{db_path}", upgrade_table=get_upgrade_table())
+        await db.start()
+        try:
+            async with db.acquire() as conn:
+                subscribed_stream = await conn.fetchrow(
+                    "SELECT * FROM streams WHERE domain=$1",
+                    "subscribed.example",
+                )
+                orphaned_stream = await conn.fetchrow(
+                    "SELECT * FROM streams WHERE domain=$1",
+                    "orphaned.example",
+                )
+        finally:
+            await db.stop()
+
+        assert subscribed_stream is not None
+        assert orphaned_stream is None
+
+    async def test_preserves_and_coerces_legacy_stream_state(
+        self, tmp_path: Path
+    ) -> None:
+        """Keep subscribed stream state while coercing legacy constrained values."""
+        db_path = tmp_path / "legacy.db"
+        with closing(sqlite3.connect(db_path)) as conn, conn:
+            conn.executescript(
+                """
+                CREATE TABLE version (version INTEGER PRIMARY KEY);
+                INSERT INTO version (version) VALUES (4);
+
+                CREATE TABLE streams (
+                    domain TEXT NOT NULL UNIQUE,
+                    name TEXT,
+                    title TEXT,
+                    online BOOLEAN NOT NULL DEFAULT false,
+                    status_since TEXT,
+                    failure_counter INTEGER DEFAULT 0,
+                    PRIMARY KEY(domain)
+                );
+
+                CREATE TABLE subscriptions (
+                    stream_domain TEXT NOT NULL,
+                    room_id TEXT NOT NULL,
+                    UNIQUE(room_id, stream_domain)
+                );
+
+                INSERT INTO streams (
+                    domain, name, title, online, status_since, failure_counter
+                )
+                VALUES
+                    (
+                        'preserved.example',
+                        'Preserved Name',
+                        'Preserved Title',
+                        1,
+                        '2026-01-01T12:00:00+00:00',
+                        7
+                    ),
+                    ('truthy-online.example', NULL, NULL, 2, NULL, 0),
+                    ('falsy-online.example', NULL, NULL, 0, NULL, 0),
+                    ('negative-counter.example', NULL, NULL, 0, NULL, -3),
+                    ('null-counter.example', NULL, NULL, 0, NULL, NULL);
+
+                INSERT INTO subscriptions (stream_domain, room_id)
+                VALUES
+                    ('preserved.example', '!preserved:example.com'),
+                    ('truthy-online.example', '!truthy:example.com'),
+                    ('falsy-online.example', '!falsy:example.com'),
+                    ('negative-counter.example', '!negative:example.com'),
+                    ('null-counter.example', '!null:example.com');
+                """
+            )
+
+        db = Database.create(f"sqlite:///{db_path}", upgrade_table=get_upgrade_table())
+        await db.start()
+        try:
+            async with db.acquire() as conn:
+                rows = await conn.fetch(
+                    """SELECT domain, name, title, online, status_since, failure_counter
+                       FROM streams
+                       ORDER BY domain"""
+                )
+        finally:
+            await db.stop()
+
+        streams = {row["domain"]: row for row in rows}
+        preserved = streams["preserved.example"]
+        assert preserved["name"] == "Preserved Name"
+        assert preserved["title"] == "Preserved Title"
+        assert preserved["online"] == 1
+        assert preserved["status_since"] == "2026-01-01T12:00:00+00:00"
+        assert preserved["failure_counter"] == 7
+        assert streams["truthy-online.example"]["online"] == 1
+        assert streams["falsy-online.example"]["online"] == 0
+        assert streams["negative-counter.example"]["failure_counter"] == 0
+        assert streams["null-counter.example"]["failure_counter"] == 0
+
+    async def test_removes_invalid_legacy_subscriptions(self, tmp_path: Path) -> None:
+        """Drop legacy subscriptions with unusable domains or room IDs."""
+        db_path = tmp_path / "legacy.db"
+        with closing(sqlite3.connect(db_path)) as conn, conn:
+            conn.executescript(
+                """
+                CREATE TABLE version (version INTEGER PRIMARY KEY);
+                INSERT INTO version (version) VALUES (4);
+
+                CREATE TABLE streams (
+                    domain TEXT NOT NULL UNIQUE,
+                    name TEXT,
+                    title TEXT,
+                    online BOOLEAN NOT NULL DEFAULT false,
+                    status_since TEXT,
+                    failure_counter INTEGER DEFAULT 0,
+                    PRIMARY KEY(domain)
+                );
+
+                CREATE TABLE subscriptions (
+                    stream_domain TEXT,
+                    room_id TEXT,
+                    UNIQUE(room_id, stream_domain)
+                );
+
+                INSERT INTO streams (domain)
+                VALUES ('valid.example'), ('invalid-room.example');
+                INSERT INTO subscriptions (stream_domain, room_id)
+                VALUES
+                    ('valid.example', '!room:example.com'),
+                    ('invalid-room.example', ''),
+                    ('', '!blank-domain:example.com'),
+                    (NULL, '!null-domain:example.com'),
+                    ('null-room.example', NULL);
+                """
+            )
+
+        db = Database.create(f"sqlite:///{db_path}", upgrade_table=get_upgrade_table())
+        await db.start()
+        try:
+            async with db.acquire() as conn:
+                streams = await conn.fetch("SELECT domain FROM streams")
+                subscriptions = await conn.fetch(
+                    "SELECT stream_domain, room_id FROM subscriptions"
+                )
+        finally:
+            await db.stop()
+
+        assert [row["domain"] for row in streams] == ["valid.example"]
+        assert [(row["stream_domain"], row["room_id"]) for row in subscriptions] == [
+            ("valid.example", "!room:example.com")
+        ]
+
 
 class TestNormalizeLegacyStatusSince:
     """Legacy timestamp normalization used by the v4 migration."""
@@ -72,35 +646,29 @@ class TestNormalizeLegacyStatusSince:
                 "2026-05-18T01:23:27+00:00",
                 id="negative-offset",
             ),
+            pytest.param(
+                datetime(2026, 5, 21, 19, 6, 24, tzinfo=UTC),
+                "2026-05-21T19:06:24+00:00",
+                id="aware-datetime",
+            ),
+            pytest.param(
+                datetime(2026, 5, 21, 19, 6, 24, tzinfo=UTC).replace(tzinfo=None),
+                None,
+                id="naive-datetime",
+            ),
             pytest.param("", None, id="blank"),
             pytest.param(None, None, id="null"),
             pytest.param("not a timestamp", None, id="malformed"),
+            pytest.param(123, None, id="unsupported-type"),
         ],
     )
     def test_normalizes_parseable_aware_timestamps(
-        self, value: str | None, expected: str | None
+        self, value: object, expected: str | None
     ) -> None:
         """Normalize parseable legacy values and ignore unusable ones."""
         assert _normalize_legacy_status_since(value) == expected
 
 
-class TestStreamExists:
-    """Stream existence checks."""
-
-    async def test_returns_true_for_existing_stream(
-        self, stream_repo: StreamRepository
-    ) -> None:
-        """Return True when the stream exists in the database."""
-        await stream_repo.create("example.com")
-        assert await stream_repo.exists("example.com") is True
-
-    async def test_returns_false_for_missing_stream(
-        self, stream_repo: StreamRepository
-    ) -> None:
-        """Return False when the stream does not exist in the database."""
-        assert await stream_repo.exists("missing.com") is False
-
-
 class TestStreamCreate:
     """Stream creation behavior."""
 
@@ -192,42 +760,34 @@ class TestStreamUpdate:
         assert state.status_since == "2026-01-01T12:00:00+00:00"
 
 
-class TestGetSubscribedStreamsForRoom:
+class TestStreamFailureCounter:
-    """Subscribed stream lookup by room."""
+    """Stream failure counter updates."""
 
-    async def test_returns_all_domains_for_room(
+    async def test_reset_failure_counter_sets_counter_to_zero(
-        self,
+        self, stream_repo: StreamRepository
-        stream_repo: StreamRepository,
-        subscription_repo: SubscriptionRepository,
     ) -> None:
-        """Return all domains a room is subscribed to."""
+        """Reset a nonzero failure counter back to zero."""
-        await stream_repo.create("alpha.com")
+        await stream_repo.create("example.com")
-        await stream_repo.create("beta.com")
+        await stream_repo.increment_failure_counter("example.com")
-        await subscription_repo.add("alpha.com", "!room1:example.com")
+        await stream_repo.increment_failure_counter("example.com")
-        await subscription_repo.add("beta.com", "!room1:example.com")
 
-        result = await subscription_repo.get_subscribed_streams_for_room(
+        await stream_repo.reset_failure_counter("example.com")
-            "!room1:example.com"
-        )
-        assert sorted(result) == ["alpha.com", "beta.com"]
 
-    async def test_returns_empty_list_for_unsubscribed_room(
+        state = await stream_repo.get_by_domain("example.com")
-        self, subscription_repo: SubscriptionRepository
+        assert state is not None
-    ) -> None:
+        assert state.failure_counter == 0
-        """Return an empty list when the room has no subscriptions."""
-        result = await subscription_repo.get_subscribed_streams_for_room(
-            "!nobody:example.com"
-        )
-        assert result == []
 
 
 class TestHasRoomSubscriptions:
     """Room subscription existence checks."""
 
     async def test_returns_true_for_subscribed_room(
-        self, subscription_repo: SubscriptionRepository
+        self,
+        stream_repo: StreamRepository,
+        subscription_repo: SubscriptionRepository,
     ) -> None:
         """Return True when the room has at least one subscription."""
+        await stream_repo.create("alpha.com")
         await subscription_repo.add("alpha.com", "!room1:example.com")
 
         assert await subscription_repo.has_room_subscriptions("!room1:example.com")
@@ -242,18 +802,17 @@ class TestHasRoomSubscriptions:
 class TestGetRoomSubscriptions:
     """Resolved room subscription lookup."""
 
-    async def test_returns_sorted_stream_states_and_skips_missing_rows(
+    async def test_returns_sorted_stream_states(
         self,
         stream_repo: StreamRepository,
         subscription_repo: SubscriptionRepository,
     ) -> None:
-        """Return sorted resolved subscriptions and skip missing stream rows."""
+        """Return sorted resolved subscriptions."""
         await stream_repo.create("beta.example")
         await stream_repo.update("beta.example", name="Beta")
         await stream_repo.create("alpha.example")
         await stream_repo.update("alpha.example", name="Alpha")
         await subscription_repo.add("beta.example", "!room:example.com")
-        await subscription_repo.add("missing.example", "!room:example.com")
         await subscription_repo.add("alpha.example", "!room:example.com")
 
         subscriptions = await subscription_repo.get_room_subscriptions(
@@ -343,7 +902,8 @@ class TestAddSubscription:
     """Subscription creation behavior."""
 
     async def test_adds_subscription(
-        self, subscription_repo: SubscriptionRepository
+        self,
+        subscription_repo: SubscriptionRepository,
     ) -> None:
         """Add a subscription row."""
         await subscription_repo.add("alpha.com", "!room1:example.com")
@@ -351,8 +911,19 @@ class TestAddSubscription:
             "!room1:example.com"
         ]
 
+    async def test_creates_stream_parent(
+        self,
+        stream_repo: StreamRepository,
+        subscription_repo: SubscriptionRepository,
+    ) -> None:
+        """Create the stream row when adding a subscription."""
+        await subscription_repo.add("alpha.com", "!room1:example.com")
+
+        assert await stream_repo.get_by_domain("alpha.com") is not None
+
     async def test_raises_when_existing(
-        self, subscription_repo: SubscriptionRepository
+        self,
+        subscription_repo: SubscriptionRepository,
     ) -> None:
         """Raise AlreadySubscribedError when a subscription already exists."""
         await subscription_repo.add("alpha.com", "!room1:example.com")
@@ -364,7 +935,8 @@ class TestRemoveSubscription:
     """Subscription removal behavior."""
 
     async def test_removes_subscription(
-        self, subscription_repo: SubscriptionRepository
+        self,
+        subscription_repo: SubscriptionRepository,
     ) -> None:
         """Remove an existing subscription row."""
         await subscription_repo.add("alpha.com", "!room1:example.com")
@@ -403,26 +975,25 @@ class TestGetAllSubscribedDomains:
         assert result == []
 
 
-class TestCountByDomain:
+class TestHasDomainSubscriptions:
-    """Subscription count by domain."""
+    """Domain subscription existence checks."""
 
-    async def test_returns_correct_count(
+    async def test_returns_true_for_subscribed_domain(
         self,
         stream_repo: StreamRepository,
         subscription_repo: SubscriptionRepository,
     ) -> None:
-        """Return the correct subscription count for a domain."""
+        """Return True when the domain has at least one subscription."""
         await stream_repo.create("alpha.com")
         await subscription_repo.add("alpha.com", "!room1:example.com")
-        await subscription_repo.add("alpha.com", "!room2:example.com")
 
-        assert await subscription_repo.count_by_domain("alpha.com") == 2
+        assert await subscription_repo.has_domain_subscriptions("alpha.com")
 
-    async def test_returns_zero_for_unknown_domain(
+    async def test_returns_false_for_unknown_domain(
         self, subscription_repo: SubscriptionRepository
     ) -> None:
-        """Return 0 for a domain with no subscriptions."""
+        """Return False when the domain has no subscriptions."""
-        assert await subscription_repo.count_by_domain("unknown.com") == 0
+        assert not await subscription_repo.has_domain_subscriptions("unknown.com")
 
 
 class TestCountByDomains:
@@ -455,29 +1026,3 @@ class TestCountByDomains:
     ) -> None:
         """Return an empty mapping when no domains are requested."""
         assert await subscription_repo.count_by_domains([]) == {}
-
-
-class TestDeleteAllForDomain:
-    """Bulk subscription deletion by domain."""
-
-    async def test_deletes_all_subscriptions_and_returns_count(
-        self,
-        stream_repo: StreamRepository,
-        subscription_repo: SubscriptionRepository,
-    ) -> None:
-        """Delete all subscriptions for the domain and return the count."""
-        await stream_repo.create("alpha.com")
-        await subscription_repo.add("alpha.com", "!room1:example.com")
-        await subscription_repo.add("alpha.com", "!room2:example.com")
-
-        deleted = await subscription_repo.delete_all_for_domain("alpha.com")
-        assert deleted == 2
-
-        rooms = await subscription_repo.get_subscribed_rooms("alpha.com")
-        assert rooms == []
-
-    async def test_returns_zero_for_unknown_domain(
-        self, subscription_repo: SubscriptionRepository
-    ) -> None:
-        """Return 0 when deleting subscriptions for an unknown domain."""
-        assert await subscription_repo.delete_all_for_domain("unknown.com") == 0

tests/test_stream_monitor.py

@@ -457,6 +457,7 @@ class TestClassifyNotification:
 
         assert notification_kind is _NotificationKind.TITLE_CHANGE
 
+
 class TestUpdateAllStreams:
     """Parallel stream update orchestration."""
 

tests/test_subscription_manager.py

@@ -54,13 +54,11 @@ def owncast_client() -> _StubOwncastClient:
 @pytest.fixture
 def manager(
     owncast_client: _StubOwncastClient,
-    stream_repo: StreamRepository,
     subscription_repo: SubscriptionRepository,
 ) -> SubscriptionManager:
     """SubscriptionManager built directly for unit tests."""
     return SubscriptionManager(
         owncast_client=owncast_client,  # type: ignore[arg-type]
-        stream_repo=stream_repo,
         subscription_repo=subscription_repo,
         logger=logging.getLogger("test"),
     )
@@ -125,7 +123,7 @@ class TestManagerSubscribe:
 
         assert domain == "stream.example"
         assert owncast_client.validated_domains == ["stream.example"]
-        assert await stream_repo.exists("stream.example") is True
+        assert await stream_repo.get_by_domain("stream.example") is not None
         assert await subscription_repo.get_subscribed_rooms("stream.example") == [
             "!room:example.com"
         ]
@@ -145,7 +143,7 @@ class TestManagerSubscribe:
 
         assert exc_info.value.domain == "bad.example"
         assert owncast_client.validated_domains == ["bad.example"]
-        assert await stream_repo.exists("bad.example") is False
+        assert await stream_repo.get_by_domain("bad.example") is None
         assert await subscription_repo.get_subscribed_rooms("bad.example") == []
 
     async def test_duplicate_subscription_raises(
@@ -193,16 +191,36 @@ class TestManagerUnsubscribe:
     async def test_removes_existing_subscription(
         self,
         manager: SubscriptionManager,
+        stream_repo: StreamRepository,
         subscription_repo: SubscriptionRepository,
     ) -> None:
-        """Existing room subscription is removed and its domain is returned."""
+        """Last subscription removal deletes the stream row."""
         await manager.subscribe("!room:example.com", "stream.example")
 
         domain = await manager.unsubscribe("!room:example.com", "stream.example")
 
         assert domain == "stream.example"
+        assert await stream_repo.get_by_domain("stream.example") is None
         assert await subscription_repo.get_subscribed_rooms("stream.example") == []
 
+    async def test_keeps_stream_for_remaining_subscriptions(
+        self,
+        manager: SubscriptionManager,
+        stream_repo: StreamRepository,
+        subscription_repo: SubscriptionRepository,
+    ) -> None:
+        """Do not delete stream state while another room is still subscribed."""
+        await manager.subscribe("!room1:example.com", "stream.example")
+        await manager.subscribe("!room2:example.com", "stream.example")
+
+        domain = await manager.unsubscribe("!room1:example.com", "stream.example")
+
+        assert domain == "stream.example"
+        assert await stream_repo.get_by_domain("stream.example") is not None
+        assert await subscription_repo.get_subscribed_rooms("stream.example") == [
+            "!room2:example.com"
+        ]
+
     async def test_missing_subscription_raises(
         self,
         manager: SubscriptionManager,
@@ -223,13 +241,12 @@ class TestManagerListings:
         stream_repo: StreamRepository,
         subscription_repo: SubscriptionRepository,
     ) -> None:
-        """Return sorted room subscriptions and skip missing stream rows."""
+        """Return sorted room subscriptions."""
         await stream_repo.create("beta.example")
         await stream_repo.update("beta.example", name="Beta")
         await stream_repo.create("alpha.example")
         await stream_repo.update("alpha.example", name="Alpha")
         await subscription_repo.add("beta.example", "!room:example.com")
-        await subscription_repo.add("missing.example", "!room:example.com")
         await subscription_repo.add("alpha.example", "!room:example.com")
 
         subscriptions = await manager.list_room_subscriptions("!room:example.com")