觸發場景：書庫管理 App 的 repository 原本是純 Future pull 介面，衍生視圖靠補償刷新（背景在 ref.watch 觀察的是 provider 圖、不是資料庫）。決策定向後要落地：repository 補 watchBooks() Stream 出口、用 StreamProvider 接進 Riverpod。 本篇範圍：落地時要答對的三個實作點，每一個的預設答案都會靜默失效。

分層落點

實作橫跨三層、每層只說自己那層的語言（歸屬判準的推導見 觀測出口的職責三分）：

契約層放介面預設實作，讓不支援觀測的實作類（測試替身、舊實作）不必立刻全部跟上：

1/// 提供書單變更的 Stream 出口，取代衍生視圖各自補償刷新。
2/// 約束：僅純 dart:async + domain entity，禁止框架型別進入此介面。
3Stream<List<Book>> watchBooks() {
4  throw UnimplementedError('watchBooks 未在此 repository 實作');
5}

實作點一：訂閱模型選 broadcast

因為書庫清單、統計頁、待補完列表都要同時觀察同一份資料，這個觀測出口有多個訂閱者。StreamController() 預設建構子是單訂閱、第二個訂閱者出現時直接 throw Bad state；這個選型的完整分析（含單訂閱在只有一個訂閱者期間完全沉默的潛伏機制）在 StreamController single vs broadcast。

1class SQLiteBookRepository implements BookRepository {
2  /// 全部寫入方法完成後透過此 controller emit 最新完整書單。
3  final StreamController<List<Book>> _booksController =
4      StreamController<List<Book>>.broadcast();
5
6  @override
7  Stream<List<Book>> watchBooks() => _booksController.stream;
8}

emit 集中在一個私有方法、掛在每個寫入方法尾端：

1Future<void> _emitCurrentBooks() async {
2  if (_booksController.isClosed) {
3    return; // repository 已 close：靜默略過，不讓通知失敗中斷寫入流程
4  }
5  final books = await getAllBooks();
6  if (!_booksController.isClosed) {
7    _booksController.add(books);
8  }
9}

兩個容易漏的細節：

1. 委派方法不重複 emit。介面上的相容性方法（saveBook 內部委派 addBook、deleteBookById 委派 deleteBook）走到底層寫入方法時已經 emit 過；在委派層再掛一次會讓一次寫入發兩次通知。emit 的掛載點是「實際執行寫入的方法集合」、不是「介面上所有看起來會寫入的方法」。
2. isClosed 要查兩次。getAllBooks() 是 async——查詢期間 repository 可能被 close，add 前不再確認就會對已關閉的 controller 拋例外，而且是從寫入方法的尾端拋出來、污染寫入本身的成功語意。

實作點二：初始值——broadcast 不補送歷史

broadcast stream 對「訂閱之前發生的事件」直接丟棄。衍生視圖訂閱 watchBooks() 的當下，上一次 emit 早就過去了——不處理初始值，畫面會停在空清單直到下一次寫入才有資料。

修法放在組裝層：StreamProvider 用 async* 先給當前值、再轉接後續變更。

1final watchBooksProvider = StreamProvider<List<Book>>((ref) async* {
2  final repository = ref.watch(bookRepositoryProvider);
3  yield await repository.getAllBooks(); // 訂閱當下：先 emit 當前完整書單
4  yield* repository.watchBooks();       // 之後：轉發 repository 的變更通知
5});

這個「當前值 + 後續變更」的組合就是 RxDart BehaviorSubject 內建的行為；純 dart:async 用兩行 yield 補上，不必為此引依賴。把初始值放組裝層而非機制層也有語意理由：repository 的 stream 誠實地只代表「變更」，「訂閱時要不要先看到當下」是消費端的呈現需求。

實作點三：dispose——關閉責任跟著 controller 的持有者

controller 的持有者是 repository，關閉責任就在 repository 的生命週期方法裡：

1Future<void> close() async {
2  await _booksController.close(); // 與資料庫連線一起釋放，避免 controller 洩漏
3  // ...既有的連線清理
4}

配合實作點一的 isClosed 防護，close 之後殘留的寫入呼叫會靜默略過通知、不會炸在使用者的操作路徑上。驗收面用三個測試釘住這組行為：寫入後 stream 收到最新書單、多訂閱者同時收到、close 後不再送出事件。

測試替身要同步這份契約

repository 有介面就有替身；替身漏掉 watchBooks() 會出現「production 正常、測試環境炸 UnimplementedError」或反過來的錯位。這次落地同步了三類替身、依「既有測試依不依賴多次 emit」給不同深度：

第三列是 Dart 特有的陷阱：mock 類別 implements 介面時不會繼承介面上的預設實作，介面加了新方法、所有 codegen mock 都要重新產生，否則消費新方法的測試在執行期才爆。mock 與真實實作的 stream 契約不對齊的後果（測試綠、production throw）在 StreamController single vs broadcast 的修復清單有完整展開。

三個必答題

把 repository stream 接給任何 reactive 框架前，三個問題各給一個明確答案：

本文範圍只涵蓋成功路徑。第四個問題——查詢失敗時 stream 該 addError 傳播還是吞掉——這裡沒有處理：_emitCurrentBooks 裡 getAllBooks() 拋例外時目前走 isClosed 防護的外圍、不會 addError，消費端收不到錯誤通知。失敗傳播的設計（要不要讓 StreamProvider 進 AsyncError 狀態、重試策略）是獨立主題；設計時要把 Riverpod 3 的預設行為算進去——失敗的 provider 會被自動重試（ProviderScope 的 retry 參數可關閉或調整間隔），addError 進到 provider 層的後果跟 2.x 不同。

介面歸屬的提醒收在最後：watchBooks() 的需求來自 Riverpod 消費端，但介面簽名只用 dart:async Stream 加 domain entity，所以它屬於 domain repository 介面——歸屬由介面用什麼語言表達決定，需求來自誰只決定介面該不該存在。Riverpod 型別止步於 watchBooksProvider、SQLite 型別止步於實作類，這條線就是三層各自的邊界。

下一步

補了觀測出口但還沒搞懂為什麼之前的導航補償和 EventBus 橋接不夠，先讀 ref.watch 觀察的是 provider 圖、不是資料庫。三層各自該放哪一層、判準怎麼來的，見 觀測出口的職責三分。單訂閱 vs broadcast 的完整選型分析在 StreamController single vs broadcast。觀測出口跟 domain event 各管什麼，見 domain event 與狀態流。