Enforced timestamps for online streams.

.gitea/workflows/cd.yml

@@ -37,7 +37,7 @@ jobs:
           grep "^version:" maubot.yaml
 
       - name: Build plugin
-        run: 7z a -tzip -mx=9 -xr!__pycache__ owncastsentry-v${{ steps.metadata.outputs.version }}.mbp owncastsentry/ maubot.yaml base-config.yaml LICENSE.txt
+        run: 7z a -tzip -mx=9 -xr!__pycache__ -xr!README.md owncastsentry-v${{ steps.metadata.outputs.version }}.mbp owncastsentry/ maubot.yaml base-config.yaml LICENSE.txt
 
       - name: Print artifact checksums
         run: |

owncastsentry/README.md

@@ -13,64 +13,46 @@ or tests.
 
 - `__init__.py`: Maubot plugin entry point, service wiring, command
   decorators, scheduler loop, shutdown, and metrics endpoint.
+- `commands.py`: Maubot command handlers for subscribe, unsubscribe,
+  subscriptions, and live listings.
 - `config.py`: Maubot configuration proxy for the Prometheus metrics endpoint
   setting.
-- `types.py`: immutable value objects, Owncast API response validation, derived
-  stream status, update results, and subscription domain errors.
-- `repository.py`: database schema migrations plus stream and subscription
-  repositories.
+- `metrics.py`: Prometheus counters, gauges, registry setup, and response timing
+  helpers.
+- `notification_service.py`: Matrix notification formatting, broadcast,
+  per-domain notification cooldowns, cleanup notices, and delivery metrics.
 - `owncast_client.py`: bounded aiohttp client for Owncast `/api/status` and
   `/api/config` requests.
-- `subscription_manager.py`: subscription use cases, domain normalization, and
-  first-subscription Owncast validation.
+- `repository.py`: database schema migrations plus stream and subscription
+  repositories.
 - `stream_monitor.py`: stream update orchestration, state transition detection,
   progressive failure backoff, cleanup thresholds, notification decisions, and
   stream metrics.
-- `notification_service.py`: Matrix notification formatting, broadcast,
-  per-domain notification cooldowns, cleanup notices, and delivery metrics.
-- `commands.py`: Maubot command handlers for subscribe, unsubscribe,
-  subscriptions, and live listings.
-- `metrics.py`: isolated Prometheus registry, counters, gauges, and response
-  timing helpers.
+- `subscription_manager.py`: subscription use cases, domain normalization, and
+  first-subscription Owncast validation.
+- `types.py`: immutable value objects, Owncast API response validation, derived
+  stream status, update results, and subscription domain errors.
 
 The package has a service-layer shape: `OwncastSentry.start()` builds the
 services, commands call `CommandHandler`, command handlers call
-`SubscriptionManager`, scheduled updates call `StreamMonitor`, and
-`StreamMonitor` coordinates repositories, the Owncast client, notifications,
-and metrics.
-
-```mermaid
-flowchart LR
-    Maubot["Maubot loads OwncastSentry"] --> Start["OwncastSentry.start()"]
-    Start --> Repos["StreamRepository and SubscriptionRepository"]
-    Start --> Owncast["OwncastClient"]
-    Start --> Metrics["MetricsService"]
-    Start --> Notify["NotificationService"]
-    Start --> Monitor["StreamMonitor"]
-    Start --> Manager["SubscriptionManager"]
-    Start --> Commands["CommandHandler"]
-    Commands --> Manager
-    Manager --> Repos
-    Manager --> Owncast
-    Monitor --> Repos
-    Monitor --> Owncast
-    Monitor --> Notify
-    Monitor --> Metrics
-    Notify --> Repos
-    Notify --> Matrix["Matrix client"]
-```
+`SubscriptionManager`, and scheduled updates call `StreamMonitor`.
+`StreamMonitor` runs the main polling workflow: it checks Owncast streams,
+applies backoff and cleanup rules, classifies state transitions, triggers any
+live/title notifications, persists changed stream state, and records stream,
+failure, and subscription metrics.
 
 ## Core Concepts
 
 A few terms are used throughout the package:
 
-- A stream is a normalized Owncast domain tracked by the plugin. The database
-  stores one stream row per domain, even if multiple Matrix rooms subscribe to
-  it.
+- A stream is a tracked [Owncast](https://owncast.online/) instance, identified
+  by its normalized bare domain. The stored entry for that domain is the stream
+  record. There is one stream record per domain, even if multiple Matrix rooms
+  subscribe to it.
 - A subscription is a Matrix room's request to receive notifications for one
   stream domain.
-- `StreamStateObservation` is one successful Owncast status API sample. The
-  monitor decides whether the observation starts a new persisted status period.
+- `StreamStateObservation` is one successful Owncast status API sample used by
+  the monitor when updating stored stream state.
 - `StreamConfigObservation` is metadata fetched from Owncast's config API,
   currently the instance name and tags used in notifications and periodic
   display-name refreshes.
@@ -83,9 +65,11 @@ Durable state lives in the database. The `streams` table stores the latest
 known display and state fields for each Owncast domain, and the `subscriptions`
 table stores which Matrix rooms follow each domain.
 
-Subscriptions are room-scoped, while stream state is shared per domain.
-Notification delivery fans out from one stored stream state to the rooms
-currently subscribed to that domain.
+Subscriptions belong to individual Matrix rooms, while stream state is shared
+per domain. Notification delivery fans out by domain to the rooms currently
+subscribed to that domain. Live/title notification text is built from the fresh
+status observation and, when available, fresh config metadata for the update
+being processed.
 
 Stream identity is the normalized bare domain. Repositories, metrics,
 notification state, and Owncast API calls all use that domain rather than the
@@ -100,41 +84,45 @@ commands, stream updates, or notifications.
 are represented by new values that are persisted through repositories instead of
 mutating existing objects.
 
-## Service Wiring
+## Plugin Boundary
+
+Maubot interacts with the package through `OwncastSentry`. That plugin class is
+the boundary between Maubot's runtime and the internal services that implement
+OwncastSentry behavior.
 
 `OwncastSentry.start()` is the composition root for one plugin instance. It
-loads config, creates the metrics service, Owncast API client, repositories,
-notification service, stream monitor, subscription manager, and command handler,
-then passes dependencies into each service explicitly.
+loads config, creates the Owncast client, repositories, metrics, notification
+service, stream monitor, subscription manager, and command handler, then passes
+dependencies into each service explicitly.
 
-This keeps lower-level modules independent of Maubot globals. Commands do not
-construct repositories, `StreamMonitor` does not know about the Maubot
-scheduler, and notification delivery does not fetch its own subscription state
-from the plugin instance.
+The plugin boundary owns Maubot integration: command registration, scheduled
+polling, and lifecycle cleanup. Scheduled ticks load subscribed domains and
+call `StreamMonitor.update_all_streams()`. `StreamMonitor` owns the per-stream
+polling workflow, while notification delivery and delivery cooldowns stay in
+`NotificationService`.
 
-The plugin boundary owns scheduling. The minute loop loads subscribed domains
-from `SubscriptionRepository` and calls `StreamMonitor.update_all_streams()`.
-The same pattern applies to new background work: schedule it from the plugin
-boundary, keep the worker service callable directly, and close owned external
-resources from `OwncastSentry.stop()`.
+Internal services do not depend on Maubot globals, which keeps command handling,
+scheduled polling, and tests wired through explicit dependencies.
 
 ## Subscription Management
 
 `SubscriptionManager` coordinates the room-to-domain relationship. Command
 handlers pass it user-supplied stream targets, and it turns those targets into
-normalized domains before touching repository state.
+normalized domains before creating or removing stored data.
 
-Subscribing a room creates a subscription for the normalized domain and creates
-the shared stream row when the domain is first seen. First-time domains are
-validated through `OwncastClient`; domains that already have subscribers reuse
-the existing stream record instead of revalidating.
+Subscribing a room validates domains with no current subscribers through
+`OwncastClient`, then inserts a subscription for the normalized domain. SQLite
+creates the shared stream record from that subscription insert when needed.
+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
-the shared stream row or make remote Owncast requests. Long-term dead stream
-cleanup is owned by `StreamMonitor`.
+Unsubscribing removes one room's subscription to a domain. When the last
+subscription for a domain is removed, SQLite deletes the shared stream record.
+Unsubscribe does not make remote Owncast requests.
 
-Listing methods stay room-scoped. They return the subscriptions relevant to one
-Matrix room, with the shared stream state attached for display.
+Listing methods return the subscriptions for one Matrix room with shared stream
+state attached, so commands can display each instance's name, title, link,
+status, and how long it has been online or offline.
 
 ## Owncast API Client
 
@@ -144,13 +132,12 @@ accepted responses into the value objects used by the rest of the package.
 
 The session uses a plugin-specific User-Agent, a dummy cookie jar, no DNS cache,
 a global connection limit, a per-host connection limit of one, and connect/read
-socket timeouts. aiohttp's own DNS cache is disabled so DNS caching and policy
-stay with the resolver configured for the deployment.
+socket timeouts. aiohttp's own DNS cache is disabled; resolver selection is left
+to aiohttp and the runtime environment.
 
 The per-host connection limit and keepalive settings are intentional. Each
-Owncast instance gets at most one reusable connection, so minute-by-minute
-status checks can reuse the existing TLS session instead of opening a fresh
-connection for every poll.
+Owncast instance is intended to keep at most one long-lived connection warm, so
+minute-by-minute status checks avoid unnecessary TLS renegotiation.
 
 Responses are accepted only when they are HTTP 200 JSON objects under the
 configured size limit. Invalid JSON, non-object JSON, oversized bodies,
@@ -163,147 +150,118 @@ for that domain.
 
 ## Stream Monitoring
 
-`StreamMonitor` owns the core per-domain state machine. `update_stream()`
-compares the stored stream state with the latest Owncast status response,
-decides whether to send a notification, updates persistent state when needed,
-and records stream metrics. Each poll first handles missing rows, polling
-backoff, and failed Owncast requests. Only a valid status response enters the
-stream transition policy. `update_all_streams()` wraps this flow for many
-domains, isolates per-domain exceptions, and records subscription-count metrics.
+`StreamMonitor` runs the package's main polling workflow for tracked Owncast
+instances. `update_stream()` owns the per-domain state machine: it compares the
+stored stream state with the latest Owncast status response, decides whether to
+send a notification, updates persistent state when needed, and records stream
+metrics. Each poll first handles stored stream lookup, polling backoff, and
+failed Owncast requests. Only a valid status response enters the stream transition
+policy. `update_all_streams()` wraps this flow for many domains, isolates
+per-domain exceptions, and records subscription-count metrics.
 
 ```mermaid
 flowchart TD
-    Start["update_stream(domain)"] --> Old["load stored StreamState"]
-    Old --> Missing{"row exists?"}
-    Missing -->|no| Noop["return success"]
-    Missing -->|yes| Backoff{"backoff allows poll?"}
-    Backoff -->|no| Skip["skip HTTP call<br/>increment failure counter"]
-    Skip --> CleanupSkip["run warning/deletion cleanup checks"]
-    CleanupSkip --> MetricsSkip["if row remains, update<br/>status and failure metrics"]
-    MetricsSkip --> SuccessSkip["return success"]
-    Backoff -->|yes| Fetch["fetch and validate /api/status"]
-    Fetch --> Fetched{"got valid state?"}
-    Fetched -->|no| Failure["increment failure counter"]
-    Failure --> CleanupFail["run warning/deletion cleanup checks"]
-    CleanupFail --> MetricsFail["if row remains, update<br/>status and failure metrics"]
-    MetricsFail --> Failed["return failure"]
-    Fetched -->|yes| Reset["reset failure counter and metric"]
-    Reset --> Timer["ensure offline timer<br/>cache entry exists"]
-    Timer --> Transition["classify stream transition"]
-    Transition --> First{"first observation?"}
-    First -->|yes| FirstLog["log suppressed notification"]
+    Start["update_stream(domain)"] --> Stored["load stored stream state"]
+    Stored --> Query{"backoff allows query?"}
+    Query -->|no| Defer["skip query for backoff"]
+    Defer --> Cleanup["apply cleanup policy"]
+    Cleanup --> Done["done"]
+    Query -->|yes| Status["fetch stream status from Owncast"]
+    Status --> Observation{"valid observation?"}
+    Observation -->|no| Failure["record failed check"]
+    Failure --> Cleanup
+    Observation -->|yes| Compare["compare observation with stored state"]
+    Compare --> First{"first observation?"}
+    First -->|yes| FirstNoNotify["no notification"]
     First -->|no| LiveCheck{"went live?"}
-    FirstLog --> ConfigDecision
-    LiveCheck -->|yes| Brief{"observed offline less than<br/>temporary cooldown?"}
-    LiveCheck -->|no| TitleChanged{"title changed<br/>while online?"}
-    Brief -->|no| LiveNotify["go-live notification needed"]
+    LiveCheck -->|yes| Brief{"offline less than 7 minutes?"}
+    Brief -->|no| LiveNotify["select go-live notification"]
     Brief -->|yes| BriefTitle{"title changed?"}
-    BriefTitle -->|yes| BriefNotify["title-change notification needed"]
-    BriefTitle -->|no| NoNotify["no notification"]
-    TitleChanged -->|yes| OfflineMarker{"offline marker newer than<br/>last notification?"}
+    BriefTitle -->|no| BriefNoNotify["no notification"]
+    BriefTitle -->|yes| TitleNotify["select title-change notification"]
+    LiveCheck -->|no| TitleChanged{"title changed while online?"}
+    TitleChanged -->|yes| TitleNotify
     TitleChanged -->|no| WentOffline{"went offline?"}
-    WentOffline -->|yes| Offline["record offline timer"]
-    WentOffline -->|no| NoNotify
-    OfflineMarker -->|yes| ResumeNotify["go-live notification needed"]
-    OfflineMarker -->|no| TitleNotify["title-change notification needed"]
-    NoNotify --> ConfigDecision
-    Offline --> ConfigDecision
-    LiveNotify --> ConfigDecision
-    BriefNotify --> ConfigDecision
-    ResumeNotify --> ConfigDecision
-    TitleNotify --> ConfigDecision
-    ConfigDecision{"notification, first observation,<br/>or hourly refresh?"}
-    ConfigDecision -->|yes| Config["fetch /api/config"]
-    ConfigDecision -->|no| NotifyGate{"notification needed?"}
-    Config --> NotifyGate
-    NotifyGate -->|yes| Send["attempt live/title notification"]
-    NotifyGate -->|no| Save["save stream row<br/>repository ignores unchanged values"]
-    Send --> Save
-    Save --> FinalMetrics["set current status metric"]
-    FinalMetrics --> Done["return success"]
+    WentOffline -->|yes| Offline["record when stream went offline"]
+    WentOffline -->|no| NoNotify["no notification"]
+    FirstNoNotify --> Metadata
+    BriefNoNotify --> Metadata
+    NoNotify --> Metadata
+    Offline --> Metadata
+    LiveNotify --> Metadata
+    TitleNotify --> Metadata
+    Metadata["fetch instance config from Owncast if display metadata is needed"]
+    Metadata --> Notify["attempt sending notification if selected"]
+    Notify --> Save["save stream update"]
+    Save --> Done
 ```
 
-Failure counters drive both status and polling behavior. A stream is displayed
-as unknown when its failure counter is above `UNKNOWN_STATUS_THRESHOLD`.
-Counters 0 through 4 poll every minute; later tiers poll only on selected
-counter values and eventually only every fifteenth minute. Skipped cycles still
-increment the counter so cleanup can progress.
+Each stream has a failure counter that tracks consecutive failed checks and
+queries skipped for backoff. The counter drives both status and polling behavior:
+a stream is displayed as unknown after more than 15 minutes of failed or
+skipped checks, and later failures progressively reduce query frequency until
+the monitor checks only every 15 minutes. Skipped queries still increment
+the counter so cleanup can progress.
 
-Transition decisions use Owncast's `online` field. The stream status timestamp is
-the bot's local UTC observation time for the first successful poll, a missing
-timestamp repair, or an online/offline transition. Owncast's remote connect,
-disconnect, and server timestamps are not required for status parsing.
+After each successful fetch from Owncast's
+[`/api/status`](https://owncast.online/api/latest/#tag/Internal/operation/GetStatus)
+endpoint, the monitor compares the stored stream state with the latest
+observation to identify transitions, such as going online, going offline, or
+changing title. Online/offline transitions are based on Owncast's `online` field.
+The stream status timestamp is the bot's UTC observation time for the first
+successful poll or an online/offline transition.
 
-First observation is special. If a new stream row has no status timestamp, the
-first successful update fetches config and stores state but suppresses
-notification attempts, even if the stream is already live.
+The first successful observation stores state and fetches config but suppresses
+notifications, even if the stream is already live. After that, only
+offline-to-online transitions can select go-live notifications; mid-session
+title changes select title-change notifications.
 
-Go-live notification attempts happen when a stream moves from offline to online
-and the monitor's in-memory offline timer shows it was not merely a brief
-outage. A stream that returns within the temporary offline cooldown only attempts
-a notification when the title changed, and then it uses a title change
-notification.
+Brief outages are treated as transient to avoid spamming Matrix rooms when an
+instance is spotty. They do not select go-live notifications. If a stream returns
+in under seven minutes with the same title, the monitor stores the new online
+state without notifying rooms. If the title changed during that brief outage, the
+monitor selects a title-change notification instead.
 
-Mid-session title changes attempt title change notifications. If the monitor has
-an offline marker that is newer than the last notification, it attempts a normal
-live notification instead so rooms do not see a title change for what is
-effectively a new session.
+Instance config is fetched from Owncast's
+[`/api/config`](https://owncast.online/api/latest/#tag/Internal/operation/GetWebConfig)
+endpoint when a live/title notification is selected, on a stream's first
+successful observation, and during the hourly refresh window for successfully
+polled streams. The config response supplies the display name and tags for
+notifications. If config is fetched successfully during an update, the persisted
+stream name is refreshed from it; otherwise the stored name is left unchanged.
 
-Offline transitions do not notify rooms. They update the database and record a
-monotonic offline timestamp used by later live/title notification policy.
-
-The live/title notification send step is an attempt, not a guaranteed Matrix message.
-`NotificationService` still applies the shared live/title cooldown and records
-the cooldown only when at least one room receives the message.
-
-`/api/config` is fetched when the monitor is about to attempt a live or
-title-change notification, on a stream's first successful observation, and
-during the hourly refresh window for successfully polled streams. The config
-response supplies the display name and tags for notifications. If config is
-fetched successfully during an update, the persisted stream name is refreshed
-from it; otherwise the stored name is left unchanged.
+Selecting a notification does not guarantee a Matrix message. The notification
+service can still suppress delivery when its per-domain live/title cooldown is
+active, which prevents repeated messages from noisy stream state changes. That
+delivery behavior is covered below.
 
 ## Cleanup Policy
 
 A domain that remains unreachable is eventually cleaned up. The failure counter
 represents minute ticks, including skipped backoff ticks.
 
-At the 83-day threshold, `StreamMonitor` sends a cleanup warning to all rooms
-subscribed to the domain. At the 90-day threshold or beyond, it sends a deletion
-notice, deletes all subscriptions for the domain, deletes the stream row, clears
-the monitor and notification caches for that domain, and removes per-domain
-metrics.
-
-```mermaid
-flowchart TD
-    Counter["failure counter after failed or skipped check"] --> Warn{"counter == 83 days?"}
-    Warn -->|yes| Warning["send cleanup warning"]
-    Warn -->|no| Delete
-    Warning --> Delete{"counter >= 90 days?"}
-    Delete -->|no| Done["done"]
-    Delete -->|yes| Notice["send cleanup deletion notice"]
-    Notice --> Subs["delete subscriptions"]
-    Subs --> Stream["delete stream row"]
-    Stream --> Caches["clear local caches"]
-    Caches --> Metrics["remove per-domain metrics"]
-    Metrics --> DoneCleanup["done"]
-```
+After 83 days of continued failed or skipped checks, `StreamMonitor` sends a
+cleanup warning to all rooms subscribed to the domain. After 90 days or more of
+continued failed or skipped checks, it sends a deletion notice, deletes all
+subscriptions for the domain, deletes the stream record, clears the monitor and
+notification caches for that domain, and asks `MetricsService` to remove
+existing per-domain metric labels. The enclosing update cycle may later recreate
+the subscription-count gauge for that domain at zero.
 
 ## Notification Delivery
 
-`NotificationService` sends plain text Matrix messages to every room subscribed
-to a domain. Broadcasts run concurrently and use `asyncio.gather(...,
-return_exceptions=True)` so one failed room does not block delivery to other
-rooms.
+`NotificationService` sends notifications to every room subscribed to a domain.
+Delivery failures are isolated per room, so one failed Matrix send does not
+block notifications to other subscribed rooms.
 
 Live and title-change notifications share a per-domain cooldown. The cooldown
 is recorded only when at least one room receives the message. If there are no
 subscribed rooms, or every delivery fails, no cooldown is recorded.
 
-Notification text is whitespace-normalized before sending. Stream names fall
-back to the domain when config lookup fails or the instance name is empty. Tags
-are appended as hash tags after filtering out empty tags and tags that start
-with a dot.
+Live/title notification text normalizes remote stream metadata before composing
+the message, so instance-provided names, titles, and tags cannot add unexpected
+line breaks or otherwise distort the notification. Messages fall back to the
+domain when a display name is unavailable.
 
 Cleanup warning and deletion notices bypass the live/title cooldown, but still
 record delivery metrics.
@@ -316,33 +274,33 @@ that catch unexpected exceptions and record command error metrics.
 
 `CommandHandler.subscribe()` and `CommandHandler.unsubscribe()` translate
 domain errors into user-facing Matrix replies. `subscriptions()` and `live()`
-read resolved room subscriptions and render Markdown output with escaped
-untrusted text.
+read room subscription listings with stream state attached and render Markdown
+output with escaped untrusted text.
 
-Command listings are room-scoped and ordered by domain through repository
-queries. `subscriptions()` includes online, offline, and unknown streams.
-`live()` includes only rows marked online with a failure count at or below the
-unknown threshold.
+Command listings are limited to the current room and ordered by domain through
+repository queries. `subscriptions()` includes online, offline, and unknown
+streams. `live()` includes only streams marked online with a failure count at or
+below the unknown threshold.
 
 ## Persistence
 
-The database schema is managed by `repository.py` through Maubot's upgrade
-table.
+Maubot owns the database connection and runs the schema upgrades registered by
+`repository.py`. The repository classes wrap that database handle and expose the
+package's stream and subscription persistence operations.
 
-- Revision 1 creates `streams` and `subscriptions`.
-- Revision 2 fixes `subscriptions.stream_domain` from `INTEGER` to `TEXT`.
-- Revision 3 adds `streams.failure_counter`.
-- Revision 4 replaces separate connect/disconnect timestamps with
-  `streams.online` and `streams.status_since`.
-
-Repositories own SQL access. `StreamRepository.update()` writes display
-and state fields; failure counters use dedicated methods.
+The persistence model stores one stream record per normalized domain and one
+subscription row per room/domain pair. SQLite constraints enforce non-empty
+identifiers, unique room/domain subscriptions, and subscription ownership by a
+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.
 
-Resolved subscription listings join `subscriptions` to `streams`, which means
-orphaned subscription rows without a stream row are skipped in room display
-queries.
+Subscription inserts create missing stream rows through a SQLite trigger, and
+deleting the last subscription for a domain deletes its stream row. Deleting a
+stream cascades to its subscriptions through SQLite, so cleanup only has to
+remove the stream record after user notifications are sent.
 
 ## Metrics
 
@@ -355,36 +313,10 @@ The registered metrics cover:
 - current stream status per domain (`1` online, `0` offline, `-1` unknown);
 - subscription count per domain;
 - consecutive check failures per domain;
-- last successful Owncast API response duration per domain;
+- successful Owncast API response duration per domain;
 - build/version information;
 - open aiohttp connection count;
 - internal scheduler and command errors.
 
 Known counter label combinations are initialized to zero. This keeps the
 Prometheus output stable before any events have occurred.
-
-## Test Map
-
-Start with these tests when changing package behavior:
-
-- `tests/test_types.py`: immutable value objects, API response shape checks,
-  truncation, stream status derivation, and subscription errors.
-- `tests/test_repository.py`: schema-backed stream and subscription repository
-  behavior.
-- `tests/test_subscription_manager.py`: domain normalization, subscribe and
-  unsubscribe workflows, validation skipping, and room listing delegation.
-- `tests/test_owncast_client.py`: Owncast API request handling, response body
-  limits, validation failures, User-Agent, response timing, and connection
-  counts.
-- `tests/test_stream_monitor.py`: polling backoff, first-update suppression,
-  live/offline/title transitions, cleanup thresholds, exception isolation, and
-  monitor metrics.
-- `tests/test_notification_service.py`: message formatting, sanitization,
-  cooldown behavior, broadcast failure accounting, cleanup notices, and
-  notification metrics.
-- `tests/test_commands.py`: end-to-end Maubot command behavior, Markdown
-  escaping, duration formatting, room-scoped subscription listings, and live
-  listings.
-- `tests/test_metrics.py`: Prometheus counters, gauges, response timer,
-  per-domain label removal, build info, open connections, and isolated registry
-  output.

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,
         )
@@ -128,7 +127,10 @@ class OwncastSentry(Plugin):
         except Exception:
             self.metrics_service.record_error(ErrorSource.COMMAND)
             self.log.exception("Unhandled exception in subscribe command.")
-            await evt.reply("An unexpected error occurred. Please try again later.")
+            await evt.reply(
+                "An unexpected error occurred. Please try again later.",
+                markdown=False,
+            )
 
     @command.new(help="Unsubscribes from an Owncast stream.")
     @command.argument("url")
@@ -139,7 +141,10 @@ class OwncastSentry(Plugin):
         except Exception:
             self.metrics_service.record_error(ErrorSource.COMMAND)
             self.log.exception("Unhandled exception in unsubscribe command.")
-            await evt.reply("An unexpected error occurred. Please try again later.")
+            await evt.reply(
+                "An unexpected error occurred. Please try again later.",
+                markdown=False,
+            )
 
     @command.new(help="Lists all stream subscriptions in this room.")
     async def subscriptions(self, evt: MessageEvent) -> None:
@@ -149,7 +154,10 @@ class OwncastSentry(Plugin):
         except Exception:
             self.metrics_service.record_error(ErrorSource.COMMAND)
             self.log.exception("Unhandled exception in subscriptions command.")
-            await evt.reply("An unexpected error occurred. Please try again later.")
+            await evt.reply(
+                "An unexpected error occurred. Please try again later.",
+                markdown=False,
+            )
 
     @command.new(help="Lists currently live streams in this room.")
     async def live(self, evt: MessageEvent) -> None:
@@ -159,7 +167,10 @@ class OwncastSentry(Plugin):
         except Exception:
             self.metrics_service.record_error(ErrorSource.COMMAND)
             self.log.exception("Unhandled exception in live command.")
-            await evt.reply("An unexpected error occurred. Please try again later.")
+            await evt.reply(
+                "An unexpected error occurred. Please try again later.",
+                markdown=False,
+            )
 
     async def _run_stream_updates_every_minute(self) -> None:
         """Run stream state updates at the top of each wall-clock minute."""

owncastsentry/commands.py

@@ -15,7 +15,7 @@
 """Command handlers for OwncastSentry bot commands."""
 
 from datetime import UTC, datetime
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, cast
 
 from .types import (
     AlreadySubscribedError,
@@ -38,9 +38,6 @@ _SECONDS_PER_DAY = 24 * _SECONDS_PER_HOUR
 
 def _sanitize_for_plain_text(text: str) -> str:
     """Sanitize text before Markdown escaping."""
-    if not text:
-        return text
-
     sanitized = text.replace("\n", " ").replace("\r", " ")
     return " ".join(sanitized.split())
 
@@ -111,18 +108,21 @@ class CommandHandler:
                 "The URL you supplied does not appear to "
                 "be a valid Owncast instance. You may have "
                 "specified an invalid domain, or the "
-                "instance is offline."
+                "instance is offline.",
+                markdown=False,
             )
             return
         except AlreadySubscribedError as e:
             await evt.reply(
-                f"This room is already subscribed to notifications for {e.domain}."
+                f"This room is already subscribed to notifications for {e.domain}.",
+                markdown=False,
             )
             return
 
         await evt.reply(
             f"Subscription added! This room will receive "
-            f"notifications when {stream_domain} goes live."
+            f"notifications when {stream_domain} goes live.",
+            markdown=False,
         )
 
     async def unsubscribe(self, evt: MessageEvent, url: str) -> None:
@@ -137,13 +137,15 @@ class CommandHandler:
             )
         except NotSubscribedError as e:
             await evt.reply(
-                f"This room is already not subscribed to notifications for {e.domain}."
+                f"This room is already not subscribed to notifications for {e.domain}.",
+                markdown=False,
             )
             return
 
         await evt.reply(
             f"Subscription removed! This room will no "
-            f"longer receive notifications for {stream_domain}."
+            f"longer receive notifications for {stream_domain}.",
+            markdown=False,
         )
 
     async def subscriptions(self, evt: MessageEvent) -> None:
@@ -185,11 +187,11 @@ class CommandHandler:
 
             # Determine status and duration (as a sub-bullet)
             match stream_state.status:
-                case StreamStatus.ONLINE if stream_state.status_since:
-                    duration = _format_duration(stream_state.status_since, now)
-                    parts.append(f"  - Status: Online for {duration}  \n")
                 case StreamStatus.ONLINE:
-                    parts.append("  - Status: Online  \n")
+                    duration = _format_duration(
+                        cast("str", stream_state.status_since), now
+                    )
+                    parts.append(f"  - Status: Online for {duration}  \n")
                 case StreamStatus.UNKNOWN:
                     parts.append("  - Status: Unknown (instance unreachable)  \n")
                 case StreamStatus.OFFLINE if stream_state.status_since:
@@ -257,11 +259,8 @@ class CommandHandler:
                 parts.append(f"  - Title: {safe_title}  \n")
 
             # Add status with duration
-            if stream_state.status_since:
-                duration = _format_duration(stream_state.status_since, now)
+            duration = _format_duration(cast("str", stream_state.status_since), now)
             parts.append(f"  - Online for {duration}  \n")
-            else:
-                parts.append("  - Online  \n")
 
             # Add stream link
             parts.append(f"  - Link: https://{domain}\n\n")

owncastsentry/notification_service.py

@@ -18,8 +18,6 @@ import asyncio
 import time
 from typing import TYPE_CHECKING, Any
 
-from mautrix.types import MessageType, TextMessageEventContent
-
 from .metrics import NotificationType
 
 if TYPE_CHECKING:
@@ -209,8 +207,7 @@ class NotificationService:
         Propagates send errors so broadcast delivery accounting can count failures.
         """
         try:
-            content = TextMessageEventContent(msgtype=MessageType.TEXT, body=body_text)
-            await self.client.send_message(room_id, content)
+            await self.client.send_notice(room_id, body_text)
         except Exception as exception:
             self.log.warning(
                 "[%s] Failed to send notification message to room [%s]: %s",

owncastsentry/repository.py

@@ -192,13 +192,248 @@ 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"""
+    )
+
+
+@upgrade_table.register(  # type: ignore[arg-type, call-arg, untyped-decorator]
+    description="Require timestamps for online streams"
+)
+async def upgrade_v6(conn: Connection) -> None:
+    """Upgrade database schema to version 6 format.
+
+    Rebuilds the stream table with a constraint that requires online streams
+    to have a status timestamp. Legacy rows that are marked online without a
+    timestamp are treated as offline because their live start time is unknown.
+
+    :param conn: A connection to run the v6 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),
+            CHECK("online" = 0 OR "status_since" IS NOT NULL),
+            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 = 1 AND status_since IS NOT NULL THEN 1
+                      ELSE 0
+                  END,
+                  status_since,
+                  failure_counter
+           FROM streams"""
+    )
+
+    await conn.execute(
+        """CREATE TABLE "subscriptions_v6_copy" (
+            "stream_domain" TEXT NOT NULL,
+            "room_id" TEXT NOT NULL,
+            PRIMARY KEY("room_id", "stream_domain")
+        )"""
+    )
+    await conn.execute(
+        """INSERT INTO subscriptions_v6_copy (stream_domain, room_id)
+           SELECT stream_domain, room_id
+           FROM subscriptions
+           WHERE EXISTS (
+               SELECT 1
+               FROM streams_new
+               WHERE streams_new.domain = subscriptions.stream_domain
+           )"""
+    )
+
+    await conn.execute('DROP TRIGGER IF EXISTS "create_stream_for_subscription"')
+    await conn.execute('DROP TRIGGER IF EXISTS "delete_unsubscribed_stream"')
+    await conn.execute('DROP INDEX IF EXISTS "subscriptions_stream_domain_idx"')
+    await conn.execute("DROP TABLE subscriptions")
+    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_v6_copy"""
+    )
+    await conn.execute("DROP TABLE subscriptions_v6_copy")
+    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 +445,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 +470,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 +532,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 +564,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 +576,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 +595,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 +608,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 +619,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]:
-        """Get all stream domains that a room is subscribed to.
-
-        :param room_id: The Matrix room ID.
-        :return: List of stream domains.
-        """
-        query = "SELECT stream_domain FROM subscriptions WHERE room_id=$1"
+    async def has_domain_subscriptions(self, domain: str) -> bool:
+        """Check whether a stream domain has any subscriptions."""
+        query = "SELECT 1 FROM subscriptions WHERE stream_domain=$1 LIMIT 1"
         async with self.db.acquire() as conn:
-            results = await conn.fetch(query, room_id)
-            return [row["stream_domain"] for row in results]
+            result = await conn.fetchrow(query, domain)
+            return result is not None
 
     async def has_room_subscriptions(self, room_id: str) -> bool:
         """Check whether a room has any subscriptions."""
@@ -419,7 +643,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 +662,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 +683,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 +693,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

@@ -134,12 +134,11 @@ def _classify_notification(
     observation: StreamStateObservation,
     *,
     offline_duration_seconds: int | None = None,
-    went_offline_after_last_notification: bool = False,
 ) -> _NotificationKind:
     """Classify the live/title notification policy for a transition.
 
-    The caller supplies timing context because it comes from the monitor's
-    monotonic offline cache and notification-service cache, not the transition.
+    The caller supplies offline duration because it comes from the monitor's
+    monotonic offline cache, not the transition.
     """
     match transition.kind:
         case _StreamTransitionKind.WENT_LIVE:
@@ -153,8 +152,6 @@ def _classify_notification(
             return _NotificationKind.LIVE
 
         case _StreamTransitionKind.TITLE_CHANGED:
-            if went_offline_after_last_notification:
-                return _NotificationKind.LIVE
             return _NotificationKind.TITLE_CHANGE
 
         case _:
@@ -314,7 +311,6 @@ class StreamMonitor:
         # Notification policy may need runtime timing context in addition to
         # the stored state and latest observation.
         offline_duration_seconds: int | None = None
-        went_offline_after_last_notification = False
 
         match transition.kind:
             case _StreamTransitionKind.FIRST_OBSERVATION:
@@ -343,11 +339,6 @@ class StreamMonitor:
                     )
             case _StreamTransitionKind.TITLE_CHANGED:
                 self.log.info("[%s] Stream title was changed!", domain)
-                # If we saw an offline event after the last notification, send
-                # a go-live notice instead of only a title-change notice.
-                went_offline_after_last_notification = self.offline_timer_cache[
-                    domain
-                ] > self.notification_service.get_last_notification_time(domain)
             case _StreamTransitionKind.WENT_OFFLINE:
                 self.offline_timer_cache[domain] = time.monotonic()
                 self.log.info("[%s] Stream is now offline.", domain)
@@ -359,7 +350,6 @@ class StreamMonitor:
             old_state,
             observation,
             offline_duration_seconds=offline_duration_seconds,
-            went_offline_after_last_notification=went_offline_after_last_notification,
         )
 
         stream_config: StreamConfigObservation | None = None
@@ -437,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)
-        if subscription_count == 0:
+        is_new_domain = not await self.subscription_repo.has_domain_subscriptions(
+            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/conftest.py

@@ -15,9 +15,12 @@
 """Shared test fixtures and stubs for OwncastSentry tests."""
 
 from dataclasses import dataclass, field
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, override
 
 import pytest
+from maubot.matrix import MaubotMatrixClient
+from maubot.testing.bot import MatrixEvent
+from mautrix.types import EventContent, EventID, EventType, RoomID
 from mautrix.util.async_db import Database
 from prometheus_client import generate_latest
 
@@ -185,27 +188,35 @@ VALID_STATUS_RESPONSE: dict[str, object] = {
 }
 
 
-@dataclass
-class _SentMessage:
-    """A message recorded by _StubMatrixClient."""
-
-    room_id: str
-    content: Any
-
-
-class _StubMatrixClient:
-    """Recording stub for the Matrix client used in notification tests."""
+class _StubMatrixClient(MaubotMatrixClient):
+    """Recording Matrix client using Mautrix message builders."""
 
     def __init__(self) -> None:
-        self.sent_messages: list[_SentMessage] = []
+        self.sent_messages: list[MatrixEvent] = []
         self.should_fail_for_rooms: set[str] = set()
 
-    async def send_message(self, room_id: str, content: Any) -> None:
-        """Record a sent message, or raise if room is in the fail set."""
+    @override
+    async def send_message_event(
+        self,
+        room_id: RoomID,
+        event_type: EventType,
+        content: EventContent,
+        _disable_encryption: bool = False,
+        **kwargs: Any,
+    ) -> EventID:
+        """Record a sent message event, or raise if room is in the fail set."""
         if room_id in self.should_fail_for_rooms:
             msg = f"Stubbed failure for room {room_id}"
             raise RuntimeError(msg)
-        self.sent_messages.append(_SentMessage(room_id=room_id, content=content))
+        self.sent_messages.append(
+            MatrixEvent(
+                room_id=room_id,
+                event_type=event_type,
+                content=content,
+                kwargs=kwargs,
+            )
+        )
+        return EventID("$stubbed")
 
 
 @dataclass

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}"
-        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 _subscribe_valid_stream(maubot_test_bot)
 
         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}"
-        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 _subscribe_valid_stream(maubot_test_bot)
 
         # 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}"
-        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 _subscribe_valid_stream(maubot_test_bot)
 
         # 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}"
-        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 _subscribe_valid_stream(maubot_test_bot)
 
         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
-        status_url = f"https://stream.logal.dev{_OWNCAST_STATUS_PATH}"
-        with aioresponses() as mocked:
-            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",
+        await _subscribe_valid_stream(maubot_test_bot)
+        await _set_stream_state(
+            maubot_plugin,
             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}"
-        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_plugin.stream_repo.update(
-            "stream.logal.dev",
+        await _subscribe_valid_stream(maubot_test_bot)
+        await _set_stream_state(
+            maubot_plugin,
             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
-        status_url = f"https://stream.logal.dev{_OWNCAST_STATUS_PATH}"
-        with aioresponses() as mocked:
-            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",
+        await _subscribe_valid_stream(maubot_test_bot)
+        await _set_stream_state(
+            maubot_plugin,
             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}"
-        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 _subscribe_valid_stream(maubot_test_bot)
 
         # 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}"
-        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 _subscribe_valid_stream(maubot_test_bot)
 
         # 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:
-            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")
+        await _subscribe_valid_streams(maubot_test_bot, "beta.com", "alpha.com")
 
-        # Set alpha online, beta offline
-        await maubot_plugin.stream_repo.update(
+        await _set_stream_state(
+            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
-        status_url = f"https://stream.logal.dev{_OWNCAST_STATUS_PATH}"
-        with aioresponses() as mocked:
-            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",
+        await _subscribe_valid_stream(maubot_test_bot)
+        await _set_stream_state(
+            maubot_plugin,
             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
-        status_url = f"https://stream.logal.dev{_OWNCAST_STATUS_PATH}"
-        with aioresponses() as mocked:
-            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",
+        await _subscribe_valid_stream(maubot_test_bot)
+        await _set_stream_state(
+            maubot_plugin,
             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:
-            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")
+        await _subscribe_valid_streams(maubot_test_bot, "beta.com", "alpha.com")
 
-        # Set both streams online with different status timestamps
-        await maubot_plugin.stream_repo.update(
+        await _set_stream_state(
+            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_notification_service.py

@@ -20,6 +20,7 @@ import time
 from typing import TYPE_CHECKING
 
 import pytest
+from mautrix.types import MessageType
 
 from owncastsentry.metrics import MetricsService
 from owncastsentry.notification_service import (
@@ -301,6 +302,7 @@ class TestNotifyStreamLive:
             "#tag"
         )
         for msg in client.sent_messages:
+            assert msg.content.msgtype == MessageType.NOTICE
             assert msg.content.body == expected_body
 
     async def test_records_cooldown_after_success(
@@ -358,10 +360,10 @@ class TestNotifyStreamLive:
         """Count a cancelled delivery result as a failure."""
         client = _StubMatrixClient()
 
-        async def send_message(_room_id: str, _content: object) -> None:
+        async def send_notice(_room_id: str, _text: str) -> None:
             raise asyncio.CancelledError
 
-        client.send_message = send_message
+        client.send_notice = send_notice
         service = _make_service(client=client, subscription_repo=subscription_repo)
 
         await stream_repo.create("example.com")
@@ -406,6 +408,7 @@ class TestNotifyStreamLive:
 
         assert len(client.sent_messages) == 1
         assert client.sent_messages[0].room_id == "!good:matrix.org"
+        assert client.sent_messages[0].content.msgtype == MessageType.NOTICE
         assert client.sent_messages[0].content.body == (
             "🎥 Stream is now live!\n"
             "Stream Title: Title\n"
@@ -432,6 +435,7 @@ class TestSendCleanupWarning:
         await service.send_cleanup_warning("example.com")
 
         assert len(client.sent_messages) == 1
+        assert client.sent_messages[0].content.msgtype == MessageType.NOTICE
         assert client.sent_messages[0].content.body == (
             "⚠️  Warning: Subscription Cleanup Scheduled\n"
             "\n"
@@ -461,6 +465,7 @@ class TestSendCleanupDeletion:
         await service.send_cleanup_deletion("example.com")
 
         assert len(client.sent_messages) == 1
+        assert client.sent_messages[0].content.msgtype == MessageType.NOTICE
         assert client.sent_messages[0].content.body == (
             "🗑️  Subscription Automatically Removed\n"
             "\n"

tests/test_owncast_client.py

@@ -477,3 +477,9 @@ class TestOpenConnectionCount:
     async def test_zero_with_no_requests(self, owncast_client: OwncastClient) -> None:
         """Return zero when no requests have been made."""
         assert owncast_client.open_connection_count == 0
+
+    async def test_zero_after_close(self, owncast_client: OwncastClient) -> None:
+        """Return zero after the HTTP session has been closed."""
+        await owncast_client.close()
+
+        assert owncast_client.open_connection_count == 0

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,716 @@ 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, online) VALUES ($1, $2)",
+                ("missing-status.example", 1),
+                id="online-without-status",
+            ),
+            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["status_since"] is None
+        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"] == 0
+        assert streams["truthy-online.example"]["status_since"] is None
+        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_v6_repairs_online_stream_without_status_timestamp(
+        self, tmp_path: Path
+    ) -> None:
+        """Repair v5 stream rows that were marked online without a timestamp."""
+        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 (5);
+
+                CREATE TABLE streams (
+                    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)
+                );
+
+                CREATE TABLE subscriptions (
+                    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
+                );
+
+                CREATE INDEX subscriptions_stream_domain_idx
+                    ON subscriptions(stream_domain);
+
+                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;
+
+                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;
+
+                INSERT INTO streams (
+                    domain, name, title, online, status_since, failure_counter
+                )
+                VALUES
+                    (
+                        'valid-live.example',
+                        'Valid Live',
+                        'Valid Title',
+                        1,
+                        '2026-01-01T12:00:00+00:00',
+                        3
+                    ),
+                    ('missing-status.example', 'Missing Status', NULL, 1, NULL, 4),
+                    ('offline.example', 'Offline', NULL, 0, NULL, 5);
+
+                INSERT INTO subscriptions (stream_domain, room_id)
+                VALUES
+                    ('valid-live.example', '!valid:example.com'),
+                    ('missing-status.example', '!missing:example.com'),
+                    ('offline.example', '!offline: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"""
+                )
+                subscriptions = await conn.fetch(
+                    """SELECT stream_domain, room_id
+                       FROM subscriptions
+                       ORDER BY stream_domain"""
+                )
+                foreign_key_errors = await conn.fetch("PRAGMA foreign_key_check")
+
+                await conn.execute(
+                    """INSERT INTO subscriptions (stream_domain, room_id)
+                       VALUES ($1, $2)""",
+                    "trigger-created.example",
+                    "!trigger:example.com",
+                )
+                trigger_created_stream = await conn.fetchrow(
+                    "SELECT * FROM streams WHERE domain=$1",
+                    "trigger-created.example",
+                )
+        finally:
+            await db.stop()
+
+        streams = {row["domain"]: row for row in rows}
+        valid_live = streams["valid-live.example"]
+        assert valid_live["online"] == 1
+        assert valid_live["status_since"] == "2026-01-01T12:00:00+00:00"
+        assert valid_live["failure_counter"] == 3
+
+        missing_status = streams["missing-status.example"]
+        assert missing_status["online"] == 0
+        assert missing_status["status_since"] is None
+        assert missing_status["failure_counter"] == 4
+
+        offline = streams["offline.example"]
+        assert offline["online"] == 0
+        assert offline["status_since"] is None
+        assert offline["failure_counter"] == 5
+
+        assert [(row["stream_domain"], row["room_id"]) for row in subscriptions] == [
+            ("missing-status.example", "!missing:example.com"),
+            ("offline.example", "!offline:example.com"),
+            ("valid-live.example", "!valid:example.com"),
+        ]
+        assert foreign_key_errors == []
+        assert trigger_created_stream is not None
+        assert trigger_created_stream["online"] == 0
+        assert trigger_created_stream["status_since"] is None
+
+    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 +786,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."""
 
@@ -191,43 +899,44 @@ class TestStreamUpdate:
         assert state.title == "Original Title"
         assert state.status_since == "2026-01-01T12:00:00+00:00"
 
-
-class TestGetSubscribedStreamsForRoom:
-    """Subscribed stream lookup by room."""
-
-    async def test_returns_all_domains_for_room(
-        self,
-        stream_repo: StreamRepository,
-        subscription_repo: SubscriptionRepository,
+    async def test_rejects_online_without_status_since(
+        self, stream_repo: StreamRepository
     ) -> None:
-        """Return all domains a room is subscribed to."""
-        await stream_repo.create("alpha.com")
-        await stream_repo.create("beta.com")
-        await subscription_repo.add("alpha.com", "!room1:example.com")
-        await subscription_repo.add("beta.com", "!room1:example.com")
+        """Reject online stream updates that leave the status timestamp unknown."""
+        await stream_repo.create("example.com")
 
-        result = await subscription_repo.get_subscribed_streams_for_room(
-            "!room1:example.com"
-        )
-        assert sorted(result) == ["alpha.com", "beta.com"]
+        with pytest.raises(sqlite3.IntegrityError):
+            await stream_repo.update("example.com", online=True)
 
-    async def test_returns_empty_list_for_unsubscribed_room(
-        self, subscription_repo: SubscriptionRepository
+
+class TestStreamFailureCounter:
+    """Stream failure counter updates."""
+
+    async def test_reset_failure_counter_sets_counter_to_zero(
+        self, stream_repo: StreamRepository
     ) -> None:
-        """Return an empty list when the room has no subscriptions."""
-        result = await subscription_repo.get_subscribed_streams_for_room(
-            "!nobody:example.com"
-        )
-        assert result == []
+        """Reset a nonzero failure counter back to zero."""
+        await stream_repo.create("example.com")
+        await stream_repo.increment_failure_counter("example.com")
+        await stream_repo.increment_failure_counter("example.com")
+
+        await stream_repo.reset_failure_counter("example.com")
+
+        state = await stream_repo.get_by_domain("example.com")
+        assert state is not None
+        assert state.failure_counter == 0
 
 
 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 +951,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 +1051,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 +1060,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 +1084,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 +1124,25 @@ class TestGetAllSubscribedDomains:
         assert result == []
 
 
-class TestCountByDomain:
-    """Subscription count by domain."""
+class TestHasDomainSubscriptions:
+    """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."""
-        assert await subscription_repo.count_by_domain("unknown.com") == 0
+        """Return False when the domain has no subscriptions."""
+        assert not await subscription_repo.has_domain_subscriptions("unknown.com")
 
 
 class TestCountByDomains:
@@ -455,29 +1175,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,31 +457,6 @@ class TestClassifyNotification:
 
         assert notification_kind is _NotificationKind.TITLE_CHANGE
 
-    def test_went_offline_after_last_notification_notifies_live(self) -> None:
-        """Notify live when stream went offline after the last notification."""
-        old_state = StreamState(
-            domain="example.com",
-            title="Old",
-            online=True,
-            status_since="2026-01-01T10:00:00+00:00",
-        )
-        observation = _stream_observation(
-            domain="example.com",
-            title="New",
-            online=True,
-            observed_at="2026-01-01T12:00:00+00:00",
-        )
-        transition = _classify_transition(old_state, observation)
-
-        notification_kind = _classify_notification(
-            transition,
-            old_state,
-            observation,
-            went_offline_after_last_notification=True,
-        )
-
-        assert notification_kind is _NotificationKind.LIVE
-
 
 class TestUpdateAllStreams:
     """Parallel stream update orchestration."""
@@ -959,12 +934,12 @@ class TestUpdateStreamTitleChange:
         )
         assert owncast.config_call_count == 1
 
-    async def test_sends_go_live_when_last_notification_before_offline(
+    async def test_sends_title_change_after_offline_gap(
         self,
         stream_repo: StreamRepository,
         subscription_repo: SubscriptionRepository,
     ) -> None:
-        """Send go-live instead of title-change after an offline gap."""
+        """Send title-change when an already-online stream changes title."""
         owncast = _StubOwncastClient(
             stream_state=_stream_observation(
                 domain="title.com",
@@ -991,8 +966,8 @@ class TestUpdateStreamTitleChange:
             status_since="2026-01-01T12:00:00+00:00",
         )
 
-        # Offline timer is MORE recent than last notification,
-        # and both are old enough to pass rate limiting
+        # Offline timer is more recent than the last notification, but title
+        # changes remain title changes once the stream is already online.
         now = time.monotonic()
         monitor.offline_timer_cache["title.com"] = (
             now - _SECONDS_BETWEEN_NOTIFICATIONS - 100
@@ -1005,7 +980,7 @@ class TestUpdateStreamTitleChange:
         assert result is True
         assert len(client.sent_messages) == 1
         assert client.sent_messages[0].content.body == (
-            "🎥 Title Stream is now live!\n"
+            "📝 Title Stream has changed its stream title!\n"
             "Stream Title: Updated Title\n"
             "\n"
             "To tune in, visit: https://title.com/"

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")