Clean up

Author: wickwirew

Sources/Feather/ConnectionPool.swift

@@ -106,7 +106,9 @@ public actor ConnectionPool: Sendable {
     private func newConnection() throws(FeatherError) -> SQLiteConnection {
         assert(count < limit)
         count += 1
-        return try SQLiteConnection(path: path)
+        let connection = try SQLiteConnection(path: path)
+        observer.installHooks(into: connection)
+        return connection
     }
 
     /// Called when we receive a connection back into the pool

Sources/Feather/DatabaseObserver.swift

@@ -8,10 +8,26 @@
 import SQLite3
 import Foundation
 
+/// A change that happened in the database.
+/// Represents many events. Changes are published
+/// after commit which may contain many events
+/// that happened within the transaction.
+public struct DatabaseChange {
+    /// A set of tables that we affected by the commit
+    public let affectedTables: Set<String>
+    /// The raw list of events from SQLite
+    public let events: [DatabaseEvent]
+}
+
+/// The raw fields SQLite gives us doing an `update_hook`
 public struct DatabaseEvent: Sendable {
+    /// What kind of operation happened
     public let operation: Operation
+    /// The database affected if any
     public let databaseName: String?
+    /// The table affected if any
     public let tableName: String?
+    /// The row id of the affected row.
     public let rowId: Int64
 
     public enum Operation: Int32, Sendable {
@@ -21,53 +37,65 @@ public struct DatabaseEvent: Sendable {
     }
 }
 
+/// Manages all of the subscriptions to a database.
+/// Will listen to SQLites hooks as well as recieve
+/// `didCommit` calls from the owning database `Connection`.
 class DatabaseObserver: @unchecked Sendable {
+    /// Lock to protect the `subscribers`.
+    /// Would have been nice to make this an actor but it would
+    /// have made things `async` that shouldn't be like
+    /// cancellation and others.
     private let lock = NSLock()
+    /// A map of all subscribers. The key is their pointer
     private var subscribers: [ObjectIdentifier: any DatabaseSubscriber] = [:]
-
+    /// We get the events from the database before the commit happens.
+    /// So if a caller requeries the database they will get old
+    /// data since the write would be in an uncommited transaction.
+    /// So we keep a list of all events that happened and then
+    /// on commit we can dispatch them all.
     private var pendingEvents: [DatabaseEvent] = []
 
+    /// Subscribes the `subscriber` to any database events.
+    /// Events are flushed upon commit and not as they come in.
     func subscribe(subscriber: any DatabaseSubscriber) {
-        lock.lock()
-        defer { lock.unlock() }
-        
-        let id = ObjectIdentifier(subscriber)
-        
-        guard subscribers[id] == nil else {
-            return
+        lock.withLock {
+            let id = ObjectIdentifier(subscriber)
+            guard subscribers[id] == nil else { return }
+            subscribers[id] = subscriber
         }
-        
-        subscribers[id] = subscriber
     }
 
+    /// Cancels the subscribers subscription.
     func cancel(subscriber: any DatabaseSubscriber) {
-        lock.lock()
-        defer { lock.unlock() }
-        
-        subscribers[ObjectIdentifier(subscriber)] = nil
-    }
-    
-    func receive(event: DatabaseEvent) {
-        lock.lock()
-        defer { lock.unlock() }
-        
-        pendingEvents.append(event)
+        lock.withLock {
+            subscribers[ObjectIdentifier(subscriber)] = nil
+        }
     }
 
+    /// Must be called by the owning database.
+    /// We do not use `sqlite3_commit_hook` since it is actually
+    /// called during the commit. Not after it.
     func didCommit() {
-        lock.lock()
-        defer { lock.unlock() }
-        
-        let events = pendingEvents
-        pendingEvents.removeAll()
-        
-        for subscriber in subscribers.values {
-            for event in events {
-                subscriber.receive(event: event)
+        lock.withLock {
+            let events = pendingEvents
+            pendingEvents.removeAll()
+            
+            // Merge all events into a single change
+            let change = DatabaseChange(
+                affectedTables: Set(events.compactMap(\.databaseName)),
+                events: events
+            )
+            
+            for subscriber in subscribers.values {
+                subscriber.receive(change: change)
             }
         }
     }
 
+    /// SQLite hooks are a per connection thing. Updates from
+    /// one connection do no publish on another so the owning
+    /// database must install the hooks for event connection
+    /// it initializes.
     func installHooks(into connection: SQLiteConnection) {
         sqlite3_update_hook(
             connection.sqliteConnection,
@@ -87,7 +115,7 @@ class DatabaseObserver: @unchecked Sendable {
         )
     }
 
-    func receiveSqliteUpdateHook(
+    private func receiveSqliteUpdateHook(
         operation: Int32,
         dbName: UnsafePointer<CChar>?,
         tableName: UnsafePointer<CChar>?,
@@ -104,10 +132,19 @@ class DatabaseObserver: @unchecked Sendable {
             rowId: rowId
         )
 
-        receive(event: event)
+        lock.withLock {
+            pendingEvents.append(event)
+        }
     }
 }
 
+
+/// A subscriber that listens to database changes
 public protocol DatabaseSubscriber: AnyObject {
-    func receive(event: DatabaseEvent)
+    /// After a commit, this is called with the `change`
+    /// which contains the events that happened during the
+    /// transaction and any additional metadata
+    ///
+    /// - Parameter change: The metadata about the change.
+    func receive(change: DatabaseChange)
 }