angd-dev/data-raft
1
0
Fork 0
Code Pull Requests Releases Activity

Refactoring

Browse Source
This commit is contained in:
Oleksii Zghurskyi
2025-11-07 20:38:09 +02:00
parent 5566f6f6ba
commit 5bbb722b20
26 changed files with 1097 additions and 770 deletions
Download Patch File Download Diff File Expand all files Collapse all files

204
Sources/DataRaft/Classes/ConnectionService.swift Normal file
View File

@@ -0,0 +1,204 @@
import Foundation
import DataLiteCore
/// A base service responsible for establishing and maintaining a database connection.
///
/// ## Overview
///
/// `ConnectionService` provides a managed execution environment for database operations. It handles
/// connection creation, configuration, encryption key application, and optional reconnection based
/// on the associated key provider’s policy.
///
/// This class guarantees thread-safe access by executing all operations within a dedicated dispatch
/// queue. Subclasses may extend it with additional behaviors, such as transaction management or
/// lifecycle event posting.
///
/// ## Topics
///
/// ### Creating a Service
///
/// - ``ConnectionProvider``
/// - ``ConnectionConfig``
/// - ``init(provider:config:queue:)``
/// - ``init(connection:config:queue:)``
///
/// ### Key Management
///
/// - ``ConnectionServiceKeyProvider``
/// - ``keyProvider``
///
/// ### Connection Lifecycle
///
/// - ``setNeedsReconnect()``
///
/// ### Performing Operations
///
/// - ``ConnectionServiceProtocol/Perform``
/// - ``perform(_:)``
open class ConnectionService:
ConnectionServiceProtocol,
@unchecked Sendable
{
// MARK: - Typealiases
/// A closure that creates a new database connection.
///
/// Used for deferred connection creation. Encapsulates initialization logic, configuration, and
/// error handling when opening the database.
///
/// - Returns: An initialized connection instance.
/// - Throws: An error if the connection cannot be created or configured.
public typealias ConnectionProvider = () throws -> ConnectionProtocol
/// A closure that configures a newly created connection.
///
/// Called after the connection is established and, if applicable, after the encryption key has
/// been applied. Use this closure to set PRAGMA options or perform additional initialization
/// logic.
///
/// - Parameter connection: The newly created connection to configure.
/// - Throws: An error if configuration fails.
public typealias ConnectionConfig = (ConnectionProtocol) throws -> Void
// MARK: - Properties
private let provider: ConnectionProvider
private let config: ConnectionConfig?
private let queue: DispatchQueue
private let queueKey = DispatchSpecificKey<Void>()
private var shouldReconnect: Bool {
keyProvider?.connectionService(shouldReconnect: self) ?? false
}
private var needsReconnect: Bool = false
private var cachedConnection: ConnectionProtocol?
private var connection: ConnectionProtocol {
get throws {
guard let cachedConnection, !needsReconnect else {
let connection = try connect()
cachedConnection = connection
needsReconnect = false
return connection
}
return cachedConnection
}
}
/// The provider responsible for supplying encryption keys to the service.
///
/// The key provider may determine whether reconnection is allowed and supply
/// the encryption key when the connection is established or restored.
public weak var keyProvider: ConnectionServiceKeyProvider?
// MARK: - Inits
/// Creates a new connection service.
///
/// Configures an internal serial queue for thread-safe access to the database. The connection
/// itself is not created during initialization — it is established lazily on first use (for
/// example, inside ``perform(_:)``).
///
/// The internal queue is created with QoS `.utility`. If `queue` is provided, it becomes the
/// target of the internal queue.
///
/// - Parameters:
/// - provider: A closure that returns a new database connection.
/// - config: An optional configuration closure called after the connection is established and
/// the encryption key is applied.
/// - queue: An optional target queue for the internal one.
public required init(
provider: @escaping ConnectionProvider,
config: ConnectionConfig? = nil,
queue: DispatchQueue? = nil
) {
self.provider = provider
self.config = config
self.queue = .init(for: Self.self, qos: .utility)
self.queue.setSpecific(key: queueKey, value: ())
if let queue = queue {
self.queue.setTarget(queue: queue)
}
}
/// Creates a new connection service using an autoclosure-based provider.
///
/// This initializer provides a convenient way to wrap an existing connection expression in an
/// autoclosure. The connection itself is not created during initialization — it is established
/// lazily on first use.
///
/// - Parameters:
/// - provider: An autoclosure that returns a new database connection.
/// - config: An optional configuration closure called after the connection is established and
/// the encryption key is applied.
/// - queue: An optional target queue for the internal one.
public required convenience init(
connection provider: @escaping @autoclosure ConnectionProvider,
config: ConnectionConfig? = nil,
queue: DispatchQueue? = nil
) {
self.init(provider: provider, config: config, queue: queue)
}
// MARK: - Connection Lifecycle
/// Marks the service as requiring reconnection before the next operation.
///
/// The reconnection behavior depends on the key provider’s implementation of
/// ``ConnectionServiceKeyProvider/connectionService(shouldReconnect:)``. If reconnection is
/// allowed, the next access to the connection will create and configure a new one.
///
/// - Returns: `true` if the reconnection flag was set; otherwise, `false`.
@discardableResult
public func setNeedsReconnect() -> Bool {
switch DispatchQueue.getSpecific(key: queueKey) {
case .none:
return queue.sync { setNeedsReconnect() }
case .some:
guard shouldReconnect else { return false }
needsReconnect = true
return true
}
}
// MARK: - Performing Operations
/// Executes a closure within the context of a managed database connection.
///
/// Runs the operation on the service’s internal queue and ensures that the connection is valid
/// before use. If the connection is unavailable or fails during execution, this method throws
/// an error.
///
/// - Parameter closure: The operation to perform using the connection.
/// - Returns: The result produced by the closure.
/// - Throws: An error thrown by the closure or the connection.
public func perform<T>(_ closure: Perform<T>) throws -> T {
switch DispatchQueue.getSpecific(key: queueKey) {
case .none: try queue.sync { try closure(connection) }
case .some: try closure(connection)
}
}
// MARK: - Internal Methods
func connect() throws -> ConnectionProtocol {
let connection = try provider()
try applyKey(to: connection)
try config?(connection)
return connection
}
func applyKey(to connection: ConnectionProtocol) throws {
guard let keyProvider = keyProvider else { return }
do {
let key = try keyProvider.connectionService(keyFor: self)
let sql = "SELECT count(*) FROM sqlite_master"
try connection.apply(key, name: nil)
try connection.execute(sql: sql)
} catch {
keyProvider.connectionService(self, didReceive: error)
throw error
}
}
}

426
Sources/DataRaft/Classes/DatabaseService.swift
View File

@@ -1,313 +1,255 @@
import Foundation
import DataLiteC
import DataLiteCore
import DataLiteC
/// Base service for working with a database.
/// A base database service handling transactions and event notifications.
///
/// `DatabaseService` provides a unified interface for performing operations using a database
/// connection, with built-in support for transactions, reconnection, and optional encryption
/// key management.
/// ## Overview
///
/// The service ensures thread-safe execution by serializing access to the connection through
/// an internal queue. This enables building modular and safe data access layers without
/// duplicating low-level logic.
/// `DatabaseService` provides a foundational layer for performing transactional database operations
/// within a thread-safe execution context. It automatically posts lifecycle notifications — such as
/// commit, rollback, and content changes — allowing observers to react to database updates in real
/// time. By default, it routes events through ``Foundation/NotificationCenter/database`` so that
/// clients can subscribe via a dedicated channel. This service is designed to be subclassed by
/// higher-level data managers that encapsulate domain logic while relying on consistent connection
/// and transaction handling.
///
/// The connection is established lazily on first use (e.g., within `perform`), not during
/// initialization. If a key provider is set, the key is applied as part of establishing or
/// restoring the connection.
///
/// Below is an example of creating a service for managing notes:
/// ## Usage
///
/// ```swift
/// final class NoteService: DatabaseService {
/// func insertNote(_ text: String) throws {
/// try perform { connection in
/// let stmt = try connection.prepare(sql: "INSERT INTO notes (text) VALUES (?)")
/// try perform(in: .deferred) { connection in
/// let sql = "INSERT INTO notes (text) VALUES (?)"
/// let stmt = try connection.prepare(sql: sql)
/// try stmt.bind(text, at: 0)
/// try stmt.step()
/// }
/// }
///
/// func fetchNotes() throws -> [String] {
/// try perform { connection in
/// let stmt = try connection.prepare(sql: "SELECT text FROM notes")
/// var result: [String] = []
/// while try stmt.step() {
/// if let text: String = stmt.columnValue(at: 0) {
/// result.append(text)
/// }
/// }
/// return result
/// }
/// }
/// }
///
/// let connection = try Connection(location: .inMemory, options: .readwrite)
/// let connection = try Connection(location: .inMemory, options: [])
/// let service = NoteService(connection: connection)
///
/// try service.insertNote("Hello, world!")
/// let notes = try service.fetchNotes()
/// print(notes) // ["Hello, world!"]
/// ```
///
/// ## Error Handling
///
/// All operations are executed on an internal serial queue, ensuring thread safety. If a
/// decryption error (`SQLITE_NOTADB`) is detected, the service may reopen the connection and
/// retry the transactional block exactly once. If the error occurs again, it is propagated
/// without further retries.
///
/// ## Encryption Key Management
///
/// If a ``keyProvider`` is set, the service uses it to obtain and apply an encryption key when
/// establishing or restoring the connection. If an error occurs while obtaining or applying the
/// key, the provider is notified through
/// ``DatabaseServiceKeyProvider/databaseService(_:didReceive:)``.
///
/// ## Reconnection
///
/// Automatic reconnection is available only during transactional blocks executed with
/// ``perform(in:closure:)``. If a decryption error (`SQLITE_NOTADB`) occurs during a
/// transaction and the provider allows reconnection, the service obtains a new key, creates a
/// new connection, and retries the block once. If the second attempt fails or reconnection is
/// disallowed, the error is propagated without further retries.
///
/// ## Topics
///
/// ### Initializers
///
/// - ``ConnectionProvider``
/// - ``ConnectionConfig``
/// - ``init(provider:config:keyProvider:queue:)``
/// - ``init(connection:config:keyProvider:queue:)``
/// - ``ConnectionService/ConnectionProvider``
/// - ``ConnectionService/ConnectionConfig``
/// - ``init(provider:config:queue:center:)``
/// - ``init(provider:config:queue:)``
///
/// ### Key Management
/// ### Performing Operations
///
/// - ``DatabaseServiceKeyProvider``
/// - ``keyProvider``
///
/// ### Database Operations
///
/// - ``DatabaseServiceProtocol/Perform``
/// - ``ConnectionServiceProtocol/Perform``
/// - ``perform(_:)``
/// - ``perform(in:closure:)``
open class DatabaseService: DatabaseServiceProtocol, @unchecked Sendable {
// MARK: - Types
/// A closure that creates a new database connection.
///
/// `ConnectionProvider` is used for deferred connection creation.
/// It allows encapsulating initialization logic, configuration, and
/// error handling when opening the database.
///
/// - Returns: An initialized `Connection` instance.
/// - Throws: An error if the connection cannot be created or configured.
public typealias ConnectionProvider = () throws -> Connection
/// A closure used to configure a newly created connection.
///
/// Called after the connection is established (and after key application if present).
/// Can be used to set PRAGMA options or perform other initialization logic.
///
/// - Parameter connection: The newly created connection.
/// - Throws: Any error if configuration fails.
public typealias ConnectionConfig = (Connection) throws -> Void
///
/// ### Connection Delegate
///
/// - ``connection(_:didUpdate:)``
/// - ``connectionWillCommit(_:)``
/// - ``connectionDidRollback(_:)``
///
/// ### Notifications
///
/// - ``databaseDidChange``
/// - ``databaseWillCommit``
/// - ``databaseDidRollback``
/// - ``databaseDidPerform``
open class DatabaseService:
ConnectionService,
DatabaseServiceProtocol,
ConnectionDelegate,
@unchecked Sendable
{
// MARK: - Properties
private let provider: ConnectionProvider
private let config: ConnectionConfig?
private let queue: DispatchQueue
private let queueKey = DispatchSpecificKey<Void>()
private let center: NotificationCenter
private var cachedConnection: Connection?
private var connection: Connection {
get throws {
guard let cachedConnection else {
let connection = try connect()
cachedConnection = connection
return connection
}
return cachedConnection
}
}
/// Encryption key provider.
/// Notification posted after the database content changes.
///
/// Used to obtain and apply a key when establishing or restoring a connection. The key is
/// requested on first access to the connection and on reconnection if needed.
public weak var keyProvider: DatabaseServiceKeyProvider?
/// Observers listen to this event to refresh cached data or update dependent components once
/// modifications are committed. The notification’s `userInfo` may include
/// ``Foundation/Notification/UserInfoKey/action`` describing the SQLite action.
public static let databaseDidChange = Notification.Name("DatabaseService.databaseDidChange")
/// Notification posted immediately before a transaction commits.
///
/// Observers can perform validation or prepare for an upcoming state change while the
/// transaction is still in progress.
public static let databaseWillCommit = Notification.Name("DatabaseService.databaseWillCommit")
/// Notification posted after a transaction rolls back.
///
/// Observers use this event to revert in-memory state or reset caches that rely on pending
/// changes.
public static let databaseDidRollback = Notification.Name("DatabaseService.databaseDidRollback")
/// Notification posted after any database operation completes, regardless of outcome.
///
/// The service emits this event after finishing a `perform(_:)` block so observers can
/// synchronize state even when the operation is read-only or aborted.
///
/// - Important: Confirm that the associated transaction was not rolled back before relying on
/// side effects.
public static let databaseDidPerform = Notification.Name("DatabaseService.databaseDidPerform")
// MARK: - Inits
/// Creates a new database service.
/// Creates a database service that posts lifecycle events to the provided notification center.
///
/// Configures the internal serial queue for thread-safe access to the database.
/// The connection is **not** created during initialization. It is established
/// lazily on first use (for example, inside `perform`).
///
/// The internal queue is always created with QoS `.utility`. If the `queue`
/// parameter is provided, it is used as the target queue for the internal one.
///
/// If a `keyProvider` is set, the encryption key will be applied when the
/// connection is established or restored.
/// The underlying connection handling matches ``ConnectionService``; the connection is created
/// lazily and all work executes on the managed serial queue.
///
/// - Parameters:
/// - provider: A closure that returns a new connection.
/// - config: An optional configuration closure called after the connection
/// is created (and after key application if present).
/// - keyProvider: An optional encryption key provider.
/// - queue: An optional target queue for the internal one.
/// - provider: A closure that returns a new database connection.
/// - config: An optional configuration closure called after the connection is established and
/// the encryption key is applied.
/// - queue: An optional target queue for the internal serial queue.
/// - center: A notification center for posting database events.
public init(
provider: @escaping ConnectionProvider,
config: ConnectionConfig? = nil,
keyProvider: DatabaseServiceKeyProvider? = nil,
queue: DispatchQueue? = nil,
center: NotificationCenter
) {
self.center = center
super.init(provider: provider, config: config, queue: queue)
}
/// Creates a database service that posts lifecycle events to the shared database notification
/// center.
///
/// The connection is established lazily on first access and all work executes on the internal
/// queue defined in ``ConnectionService``.
///
/// - Parameters:
/// - provider: A closure that returns a new database connection.
/// - config: An optional configuration closure called after the connection is established and
/// the encryption key is applied.
/// - queue: An optional target queue for the internal serial queue.
public required init(
provider: @escaping ConnectionProvider,
config: ConnectionConfig? = nil,
queue: DispatchQueue? = nil
) {
self.provider = provider
self.config = config
self.keyProvider = keyProvider
self.queue = .init(for: Self.self, qos: .utility)
self.queue.setSpecific(key: queueKey, value: ())
if let queue = queue {
self.queue.setTarget(queue: queue)
self.center = .database
super.init(provider: provider, config: config, queue: queue)
}
// MARK: - Performing Operations
/// Executes a closure with a managed database connection and posts a completion notification.
///
/// The override mirrors ``ConnectionService/perform(_:)`` for queue-confined execution while
/// ensuring ``DatabaseService/databaseDidPerform`` is delivered after the closure completes.
///
/// - Parameter closure: The operation to execute using the open connection.
/// - Returns: The value returned by the closure.
/// - Throws: Errors thrown by the closure or underlying connection.
public override func perform<T>(_ closure: Perform<T>) throws -> T {
try super.perform { connection in
defer { center.post(name: Self.databaseDidPerform, object: self) }
return try closure(connection)
}
}
/// Creates a new database service.
/// Executes a closure inside a transaction when the connection operates in autocommit mode.
///
/// The connection is created lazily on first use. If a `keyProvider` is set,
/// the key will be applied when the connection is established.
/// The method begins the requested `TransactionType`, runs the closure, and commits the
/// transaction on success. Failures trigger a rollback. If the SQLite engine reports
/// `SQLITE_NOTADB` and the key provider allows reconnection, the service re-establishes the
/// connection and retries the closure once, mirroring the behavior described in
/// ``DatabaseServiceProtocol``.
///
/// - Parameters:
/// - provider: An expression that creates a new connection.
/// - config: An optional configuration closure called after the connection
/// is created (and after key application if present).
/// - keyProvider: An optional encryption key provider.
/// - queue: An optional target queue for the internal one.
public convenience init(
connection provider: @escaping @autoclosure ConnectionProvider,
config: ConnectionConfig? = nil,
keyProvider: DatabaseServiceKeyProvider? = nil,
queue: DispatchQueue? = nil
) {
self.init(
provider: provider,
config: config,
keyProvider: keyProvider,
queue: queue
)
}
// MARK: - Methods
/// Executes a closure with the current connection.
///
/// Ensures thread-safe access by running the closure on the internal serial queue.
/// The connection is created lazily if needed.
///
/// - Parameter closure: A closure that takes the active connection.
/// - transaction: The type of transaction to start (for example, `.deferred`).
/// - closure: The work to run while the transaction is active.
/// - Returns: The value returned by the closure.
/// - Throws: An error if the connection cannot be created or if the closure throws.
final public func perform<T>(_ closure: Perform<T>) throws -> T {
try withConnection(closure)
}
/// Executes a closure inside a transaction if the connection is in autocommit mode.
/// - Throws: Errors from the closure, transaction handling, or connection management.
///
/// If the connection is in autocommit mode, starts a new transaction of the specified
/// type, executes the closure, and commits changes on success. If the closure throws
/// an error, the transaction is rolled back.
///
/// If the closure throws `Connection.Error` with code `SQLITE_NOTADB` and reconnection
/// is allowed, the service attempts to create a new connection, reapply the key, and
/// retries the transaction block once. If the second attempt fails or reconnection
/// is disallowed, the error is propagated without further retries.
///
/// If a transaction is already active (connection not in autocommit mode), the closure
/// is executed directly without starting a new transaction.
///
/// - Parameters:
/// - transaction: The type of transaction to start.
/// - closure: A closure that takes the active connection and returns a result.
/// - Returns: The value returned by the closure.
/// - Throws: Errors from connection creation, key application, configuration,
/// transaction management, or from the closure itself.
/// - Important: The closure may be executed more than once. Ensure it is idempotent.
final public func perform<T>(
/// - Important: The closure may be executed more than once if a reconnection occurs. Ensure it
/// performs only database operations and does not produce external side effects (such as
/// sending network requests or posting notifications).
public func perform<T>(
in transaction: TransactionType,
closure: Perform<T>
) throws -> T {
try withConnection { connection in
if connection.isAutocommit {
do {
try connection.beginTransaction(transaction)
let result = try closure(connection)
try connection.commitTransaction()
return result
} catch {
try perform { connection in
guard connection.isAutocommit else {
return try closure(connection)
}
do {
try connection.beginTransaction(transaction)
let result = try closure(connection)
try connection.commitTransaction()
return result
} catch {
if !connection.isAutocommit {
try connection.rollbackTransaction()
guard let error = error as? Connection.Error,
error.code == SQLITE_NOTADB,
shouldReconnect
else { throw error }
try reconnect()
return try withConnection { connection in
do {
try connection.beginTransaction(transaction)
let result = try closure(connection)
try connection.commitTransaction()
return result
} catch {
}
guard
let error = error as? SQLiteError,
error.code == SQLITE_NOTADB,
setNeedsReconnect()
else {
throw error
}
return try perform { connection in
do {
try connection.beginTransaction(transaction)
let result = try closure(connection)
try connection.commitTransaction()
return result
} catch {
if !connection.isAutocommit {
try connection.rollbackTransaction()
throw error
}
throw error
}
}
} else {
return try closure(connection)
}
}
}
}
// MARK: - Private
private extension DatabaseService {
var shouldReconnect: Bool {
keyProvider?.databaseService(shouldReconnect: self) ?? false
// MARK: - ConnectionDelegate
/// Posts ``DatabaseService/databaseDidChange`` when the database content updates.
///
/// - Parameters:
/// - connection: The connection that performed the change.
/// - action: The SQLite action describing the modification.
public func connection(_ connection: any ConnectionProtocol, didUpdate action: SQLiteAction) {
let userInfo = [Notification.UserInfoKey.action: action]
center.post(name: Self.databaseDidChange, object: self, userInfo: userInfo)
}
func withConnection<T>(_ closure: Perform<T>) throws -> T {
switch DispatchQueue.getSpecific(key: queueKey) {
case .none: try queue.asyncAndWait { try closure(connection) }
case .some: try closure(connection)
}
/// Posts ``DatabaseService/databaseWillCommit`` before a transaction commits.
///
/// - Parameter connection: The connection preparing to commit.
public func connectionWillCommit(_ connection: any ConnectionProtocol) throws {
center.post(name: Self.databaseWillCommit, object: self)
}
func reconnect() throws {
cachedConnection = try connect()
/// Posts ``DatabaseService/databaseDidRollback`` after a transaction rollback.
///
/// - Parameter connection: The connection that rolled back.
public func connectionDidRollback(_ connection: any ConnectionProtocol) {
center.post(name: Self.databaseDidRollback, object: self)
}
func connect() throws -> Connection {
let connection = try provider()
try applyKey(to: connection)
try config?(connection)
// MARK: - Internal Methods
override func connect() throws -> any ConnectionProtocol {
let connection = try super.connect()
connection.add(delegate: self)
return connection
}
func applyKey(to connection: Connection) throws {
guard let keyProvider = keyProvider else { return }
do {
let key = try keyProvider.databaseService(keyFor: self)
let sql = "SELECT count(*) FROM sqlite_master"
try connection.apply(key)
try connection.execute(raw: sql)
} catch {
keyProvider.databaseService(self, didReceive: error)
throw error
}
}
}

246
Sources/DataRaft/Classes/MigrationService.swift
View File

@@ -1,135 +1,195 @@
import Foundation
import DataLiteCore
/// Thread-safe service for executing ordered database schema migrations.
#if os(Windows)
import WinSDK
#endif
/// A service that executes ordered database schema migrations.
///
/// `MigrationService` stores registered migrations and applies them sequentially
/// to update the database schema. Each migration runs only once, in version order,
/// based on the current schema version stored in the database.
/// ## Overview
///
/// The service is generic over:
/// - `Service`: a database service conforming to ``DatabaseServiceProtocol``
/// - `Storage`: a version storage conforming to ``VersionStorage``
/// This class manages migration registration and applies them sequentially to update the database
/// schema. Each migration corresponds to a specific version and runs only once, ensuring that
/// schema upgrades are applied in a consistent, deterministic way.
///
/// Migrations are identified by version and script URL. Both must be unique
/// across all registered migrations.
/// Migrations are executed within an exclusive transaction — if any step fails, the entire process
/// is rolled back, leaving the database unchanged.
///
/// Execution is performed inside a single `.exclusive` transaction, ensuring
/// that either all pending migrations are applied successfully or none are.
/// On error, the database state is rolled back to the original version.
/// `MigrationService` coordinates the migration process by:
/// - Managing a registry of unique migrations.
/// - Reading and writing the current schema version through a ``VersionStorage`` implementation.
/// - Executing SQL scripts in ascending version order.
///
/// This type is safe to use from multiple threads.
/// It is safe for concurrent use. Internally, it uses a POSIX mutex to ensure thread-safe
/// registration and execution.
///
/// ## Usage
///
/// ```swift
/// let connection = try Connection(location: .inMemory, options: .readwrite)
/// let storage = UserVersionStorage<BitPackVersion>()
/// let service = MigrationService(service: connectionService, storage: storage)
/// let storage = UserVersionStorage<SemanticVersion>()
/// let service = MigrationService(provider: { connection }, storage: storage)
///
/// try service.add(Migration(version: "1.0.0", byResource: "v_1_0_0.sql")!)
/// try service.add(Migration(version: "1.0.1", byResource: "v_1_0_1.sql")!)
/// try service.add(Migration(version: "1.0.0", byResource: "v1_0_0.sql")!)
/// try service.add(Migration(version: "1.0.1", byResource: "v1_0_1.sql")!)
/// try service.migrate()
/// ```
///
/// ### Custom Versions and Storage
/// ## Topics
///
/// You can supply a custom `Version` type conforming to ``VersionRepresentable``
/// and a `VersionStorage` implementation that determines how and where the
/// version is persisted (e.g., `PRAGMA user_version`, metadata table, etc.).
/// ### Initializers
///
/// - ``init(provider:config:queue:storage:)``
/// - ``init(provider:config:queue:)``
///
/// ### Migration Management
///
/// - ``add(_:)``
/// - ``migrate()``
public final class MigrationService<
Service: DatabaseServiceProtocol,
Storage: VersionStorage
>:
ConnectionService,
MigrationServiceProtocol,
@unchecked Sendable
{
/// Schema version type used for migration ordering.
// MARK: - Typealiases
/// The type representing schema version ordering.
public typealias Version = Storage.Version
private let service: Service
// MARK: - Properties
private let storage: Storage
private var mutex = pthread_mutex_t()
private var migrations = Set<Migration<Version>>()
/// Encryption key provider delegated to the underlying database service.
public weak var keyProvider: DatabaseServiceKeyProvider? {
get { service.keyProvider }
set { service.keyProvider = newValue }
}
#if os(Windows)
private var mutex = SRWLOCK()
#else
private var mutex = pthread_mutex_t()
#endif
/// Creates a migration service with the given database service and storage.
// MARK: - Inits
/// Creates a migration service with a specified connection configuration and version storage.
///
/// - Parameters:
/// - service: Database service used to execute migrations.
/// - storage: Version storage for reading and writing schema version.
/// - provider: A closure that returns a new database connection.
/// - config: An optional configuration closure called after the connection is established.
/// - queue: An optional target queue for internal database operations.
/// - storage: The version storage responsible for reading and writing schema version data.
public init(
service: Service,
provider: @escaping ConnectionProvider,
config: ConnectionConfig? = nil,
queue: DispatchQueue? = nil,
storage: Storage
) {
self.service = service
self.storage = storage
pthread_mutex_init(&mutex, nil)
super.init(provider: provider, config: config, queue: queue)
#if os(Windows)
InitializeSRWLock(&mutex)
#else
pthread_mutex_init(&mutex, nil)
#endif
}
/// Creates a migration service using the default version storage.
///
/// - Parameters:
/// - provider: A closure that returns a new database connection.
/// - config: An optional configuration closure called after the connection is established.
/// - queue: An optional target queue for internal database operations.
public required init(
provider: @escaping ConnectionProvider,
config: ConnectionConfig? = nil,
queue: DispatchQueue? = nil
) {
self.storage = .init()
super.init(provider: provider, config: config, queue: queue)
#if os(Windows)
InitializeSRWLock(&mutex)
#else
pthread_mutex_init(&mutex, nil)
#endif
}
deinit {
pthread_mutex_destroy(&mutex)
#if !os(Windows)
pthread_mutex_destroy(&mutex)
#endif
}
// MARK: - Unsupported
@available(*, unavailable)
public override func setNeedsReconnect() -> Bool {
fatalError("Reconnection is not supported for MigrationService.")
}
@available(*, unavailable)
public override func perform<T>(_ closure: Perform<T>) throws -> T {
fatalError("Direct perform is not supported for MigrationService.")
}
// MARK: - Migration Management
/// Registers a new migration, ensuring version and script URL uniqueness.
///
/// - Parameter migration: The migration to register.
/// - Throws: ``MigrationError/duplicateMigration(_:)`` if the migration's
/// version or script URL is already registered.
/// - Throws: ``MigrationError/duplicateMigration(_:)`` if a migration with the same version or
/// script URL is already registered.
public func add(_ migration: Migration<Version>) throws(MigrationError<Version>) {
pthread_mutex_lock(&mutex)
defer { pthread_mutex_unlock(&mutex) }
guard !migrations.contains(where: {
$0.version == migration.version
|| $0.scriptURL == migration.scriptURL
}) else {
#if os(Windows)
AcquireSRWLockExclusive(&mutex)
defer { ReleaseSRWLockExclusive(&mutex) }
#else
pthread_mutex_lock(&mutex)
defer { pthread_mutex_unlock(&mutex) }
#endif
guard
!migrations.contains(where: {
$0.version == migration.version
|| $0.scriptURL == migration.scriptURL
})
else {
throw .duplicateMigration(migration)
}
migrations.insert(migration)
}
/// Executes all pending migrations inside a single exclusive transaction.
/// Executes all pending migrations in ascending version order.
///
/// This method retrieves the current schema version from storage, then determines
/// which migrations have a higher version. The selected migrations are sorted in
/// ascending order and each one's SQL script is executed in sequence. When all
/// scripts complete successfully, the stored version is updated to the highest
/// applied migration.
/// The service retrieves the current version from ``VersionStorage``, selects migrations with
/// higher versions, sorts them, and executes their scripts inside an exclusive transaction.
///
/// If a script is empty or execution fails, the process aborts and the transaction
/// is rolled back, leaving the database unchanged.
///
/// - Throws: ``MigrationError/emptyMigrationScript(_:)`` if a script is empty.
/// - Throws: ``MigrationError/migrationFailed(_:_:)`` if execution or version
/// update fails.
/// - Throws: ``MigrationError/emptyMigrationScript(_:)`` if a migration script is empty.
/// - Throws: ``MigrationError/migrationFailed(_:_:)`` if a migration fails to execute or the
/// version update cannot be persisted.
public func migrate() throws(MigrationError<Version>) {
pthread_mutex_lock(&mutex)
defer { pthread_mutex_unlock(&mutex) }
#if os(Windows)
AcquireSRWLockExclusive(&mutex)
defer { ReleaseSRWLockExclusive(&mutex) }
#else
pthread_mutex_lock(&mutex)
defer { pthread_mutex_unlock(&mutex) }
#endif
do {
try service.perform(in: .exclusive) { connection in
try storage.prepare(connection)
let version = try storage.getVersion(connection)
let migrations = migrations
.filter { $0.version > version }
.sorted { $0.version < $1.version }
for migration in migrations {
let script = try migration.script
guard !script.isEmpty else {
throw MigrationError.emptyMigrationScript(migration)
try super.perform { connection in
do {
try connection.beginTransaction(.exclusive)
try migrate(with: connection)
try connection.commitTransaction()
} catch {
if !connection.isAutocommit {
try connection.rollbackTransaction()
}
do {
try connection.execute(sql: script)
} catch {
throw MigrationError.migrationFailed(migration, error)
}
}
if let version = migrations.last?.version {
try storage.setVersion(connection, version)
throw error
}
}
} catch let error as MigrationError<Version> {
@@ -139,3 +199,31 @@ public final class MigrationService<
}
}
}
// MARK: - Private
private extension MigrationService {
func migrate(with connection: ConnectionProtocol) throws {
try storage.prepare(connection)
let version = try storage.getVersion(connection)
let migrations = migrations
.filter { $0.version > version }
.sorted { $0.version < $1.version }
for migration in migrations {
let script = try migration.script
guard !script.isEmpty else {
throw MigrationError.emptyMigrationScript(migration)
}
do {
try connection.execute(sql: script)
} catch {
throw MigrationError.migrationFailed(migration, error)
}
}
if let version = migrations.last?.version {
try storage.setVersion(connection, version)
}
}
}

138
Sources/DataRaft/Classes/ModelDatabaseService.swift Normal file
View File

@@ -0,0 +1,138 @@
import Foundation
import DataLiteCoder
/// A database service that provides model encoding and decoding support.
///
/// ## Overview
///
/// `ModelDatabaseService` extends ``DatabaseService`` by integrating `RowEncoder` and `RowDecoder`
/// to simplify model-based interactions with the database. Subclasses can encode Swift types into
/// SQLite rows and decode query results back into strongly typed models.
///
/// This enables a clean, type-safe persistence layer for applications that use Codable or custom
/// encodable/decodable types.
///
/// `ModelDatabaseService` serves as a foundation for higher-level model repositories and services.
/// It inherits all transactional and thread-safe behavior from ``DatabaseService`` while adding
/// automatic model serialization.
///
/// ## Usage
///
/// ```swift
/// struct User: Codable {
/// let id: Int
/// let name: String
/// }
///
/// final class UserService: ModelDatabaseService, @unchecked Sendable {
/// func fetchUser() throws -> User? {
/// try perform(in: .deferred) { connection in
/// let stmt = try connection.prepare(sql: "SELECT * FROM users")
/// guard try stmt.step(), let row = stmt.currentRow() else {
/// return nil
/// }
/// return try decoder.decode(User.self, from: row)
/// }
/// }
///
/// func insertUser(_ user: User) throws {
/// try perform(in: .immediate) { connection in
/// let row = try encoder.encode(user)
/// let columns = row.columns.joined(separator: ", ")
/// let placeholders = row.namedParameters.joined(separator: ", ")
/// let sql = "INSERT INTO users (\(columns)) VALUES (\(placeholders))"
/// let stmt = try connection.prepare(sql: sql)
/// try stmt.execute([row])
/// }
/// }
/// }
/// ```
///
/// ## Topics
///
/// ### Properties
///
/// - ``encoder``
/// - ``decoder``
///
/// ### Initializers
///
/// - ``init(provider:config:queue:center:encoder:decoder:)``
/// - ``init(provider:config:queue:)``
/// - ``init(connection:config:queue:)``
open class ModelDatabaseService: DatabaseService, @unchecked Sendable {
// MARK: - Properties
/// The encoder used to serialize models into row representations.
public let encoder: RowEncoder
/// The decoder used to deserialize database rows into model instances.
public let decoder: RowDecoder
// MARK: - Inits
/// Creates a model-aware database service.
///
/// - Parameters:
/// - provider: A closure that returns a new database connection.
/// - config: Optional configuration for the connection.
/// - queue: The dispatch queue used for serializing database operations.
/// - center: The notification center used for database events. Defaults to `.database`.
/// - encoder: The encoder for converting models into SQLite rows.
/// - decoder: The decoder for converting rows back into model instances.
public init(
provider: @escaping ConnectionProvider,
config: ConnectionConfig? = nil,
queue: DispatchQueue? = nil,
center: NotificationCenter = .database,
encoder: RowEncoder,
decoder: RowDecoder
) {
self.encoder = encoder
self.decoder = decoder
super.init(
provider: provider,
config: config,
queue: queue,
center: center
)
}
/// Creates a model-aware database service using default encoder and decoder instances.
///
/// - Parameters:
/// - provider: A closure that returns a new database connection.
/// - config: Optional configuration for the connection.
/// - queue: The dispatch queue used for serializing database operations.
public required init(
provider: @escaping ConnectionProvider,
config: ConnectionConfig? = nil,
queue: DispatchQueue? = nil
) {
self.encoder = .init()
self.decoder = .init()
super.init(
provider: provider,
config: config,
queue: queue
)
}
/// Creates a model-aware database service from a connection autoclosure.
///
/// - Parameters:
/// - provider: A connection autoclosure that returns a database connection.
/// - config: Optional configuration for the connection.
/// - queue: The dispatch queue used for serializing database operations.
public required convenience init(
connection provider: @escaping @autoclosure ConnectionProvider,
config: ConnectionConfig? = nil,
queue: DispatchQueue? = nil
) {
self.init(
provider: provider,
config: config,
queue: queue
)
}
}

115
Sources/DataRaft/Classes/RowDatabaseService.swift
View File

@@ -1,115 +0,0 @@
import Foundation
import DataLiteCore
import DataLiteCoder
/// A database service that provides built-in row encoding and decoding.
///
/// `RowDatabaseService` extends `DatabaseService` by adding support for
/// value serialization using `RowEncoder` and deserialization using `RowDecoder`.
///
/// This enables subclasses to perform type-safe operations on models
/// encoded from or decoded into SQLite row representations.
///
/// For example, a concrete service might define model-aware fetch or insert methods:
///
/// ```swift
/// struct User: Codable {
/// let id: Int
/// let name: String
/// }
///
/// final class UserService: RowDatabaseService {
/// func fetchUsers() throws -> [User] {
/// try perform(in: .deferred) { connection in
/// let stmt = try connection.prepare(sql: "SELECT * FROM users")
/// let rows = try stmt.execute()
/// return try decoder.decode([User].self, from: rows)
/// }
/// }
///
/// func insertUser(_ user: User) throws {
/// try perform(in: .deferred) { connection in
/// let row = try encoder.encode(user)
/// let columns = row.columns.joined(separator: ", ")
/// let parameters = row.namedParameters.joined(separator: ", ")
/// let stmt = try connection.prepare(
/// sql: "INSERT INTO users (\(columns)) VALUES (\(parameters))"
/// )
/// try stmt.execute(rows: [row])
/// }
/// }
/// }
/// ```
///
/// `RowDatabaseService` encourages a reusable, type-safe pattern for
/// model-based interaction with SQLite while preserving thread safety
/// and transactional integrity.
open class RowDatabaseService:
DatabaseService,
RowDatabaseServiceProtocol,
@unchecked Sendable
{
// MARK: - Properties
/// The encoder used to serialize values into row representations.
public let encoder: RowEncoder
/// The decoder used to deserialize row values into strongly typed models.
public let decoder: RowDecoder
// MARK: - Inits
/// Creates a new `RowDatabaseService`.
///
/// This initializer accepts a closure that supplies the database connection. If no encoder
/// or decoder is provided, default instances are used.
///
/// - Parameters:
/// - provider: A closure that returns a `Connection` instance. May throw an error.
/// - encoder: The encoder used to serialize models into SQLite-compatible rows.
/// Defaults to a new encoder.
/// - decoder: The decoder used to deserialize SQLite rows into typed models.
/// Defaults to a new decoder.
/// - queue: An optional dispatch queue used for serialization. If `nil`, an internal
/// serial queue with `.utility` QoS is created.
/// - Throws: Any error thrown by the connection provider.
public convenience init(
connection provider: @escaping @autoclosure ConnectionProvider,
encoder: RowEncoder = RowEncoder(),
decoder: RowDecoder = RowDecoder(),
queue: DispatchQueue? = nil
) {
self.init(
provider: provider,
encoder: encoder,
decoder: decoder,
queue: queue
)
}
/// Designated initializer for `RowDatabaseService`.
///
/// Initializes a new instance with the specified connection provider, encoder, decoder,
/// and an optional dispatch queue for synchronization.
///
/// - Parameters:
/// - provider: A closure that returns a `Connection` instance. May throw an error.
/// - encoder: A custom `RowEncoder` used for encoding model data. Defaults to a new encoder.
/// - decoder: A custom `RowDecoder` used for decoding database rows. Defaults to a new decoder.
/// - queue: An optional dispatch queue for serializing access to the database connection.
/// If `nil`, a default internal serial queue with `.utility` QoS is used.
/// - Throws: Any error thrown by the connection provider.
public init(
provider: @escaping ConnectionProvider,
encoder: RowEncoder = RowEncoder(),
decoder: RowDecoder = RowDecoder(),
queue: DispatchQueue? = nil
) {
self.encoder = encoder
self.decoder = decoder
super.init(
provider: provider,
queue: queue
)
}
}

75
Sources/DataRaft/Classes/UserVersionStorage.swift
View File

@@ -1,45 +1,57 @@
import Foundation
import DataLiteCore
/// A database version storage that uses the `user_version` field.
/// A version storage that persists schema versions in SQLite’s `user_version` field.
///
/// This class implements ``VersionStorage`` by storing version information
/// in the SQLite `PRAGMA user_version` field. It provides a lightweight,
/// type-safe way to persist versioning data in a database.
/// ## Overview
///
/// The generic `Version` type must conform to both ``VersionRepresentable``
/// and `RawRepresentable`, where `RawValue == UInt32`. This allows
/// converting between stored integer values and semantic version types
/// defined by the application.
/// `UserVersionStorage` provides a lightweight, type-safe implementation of ``VersionStorage`` that
/// stores version data using the SQLite `PRAGMA user_version` mechanism. This approach is simple,
/// efficient, and requires no additional tables.
///
/// The generic ``Version`` type must conform to both ``VersionRepresentable`` and
/// `RawRepresentable`, with `RawValue == UInt32`. This enables conversion between stored integer
/// values and the application’s semantic version type.
///
/// ## Topics
///
/// ### Errors
///
/// - ``Error``
///
/// ### Instance Methods
///
/// - ``getVersion(_:)``
/// - ``setVersion(_:_:)``
public final class UserVersionStorage<
Version: VersionRepresentable & RawRepresentable
>: Sendable, VersionStorage where Version.RawValue == UInt32 {
/// Errors related to reading or decoding the version.
/// Errors related to reading or decoding the stored version.
public enum Error: Swift.Error {
/// The stored `user_version` could not be decoded into a valid `Version` case.
/// The stored `user_version` value could not be decoded into a valid ``Version``.
///
/// - Parameter value: The invalid raw `UInt32` value.
case invalidStoredVersion(UInt32)
}
// MARK: - Inits
/// Creates a new user version storage instance.
/// Creates a new instance of user version storage.
public init() {}
// MARK: - Methods
// MARK: - Version Management
/// Returns the current version stored in the `user_version` field.
/// Returns the current schema version stored in the `user_version` field.
///
/// This method reads the `PRAGMA user_version` value and attempts to
/// decode it into a valid `Version` value. If the stored value is not
/// recognized, it throws an error.
/// Reads the `PRAGMA user_version` value and attempts to decode it into a valid ``Version``.
/// If decoding fails, this method throws an error.
///
/// - Parameter connection: The database connection.
/// - Returns: A decoded version value of type `Version`.
/// - Throws: ``Error/invalidStoredVersion(_:)`` if the stored value
/// cannot be mapped to a valid `Version` instance.
public func getVersion(
_ connection: Connection
) throws -> Version {
/// - Parameter connection: The active database connection.
/// - Returns: The decoded version value.
/// - Throws: ``Error/invalidStoredVersion(_:)`` if the stored value cannot
/// be mapped to a valid version case.
public func getVersion(_ connection: ConnectionProtocol) throws -> Version {
let raw = UInt32(bitPattern: connection.userVersion)
guard let version = Version(rawValue: raw) else {
throw Error.invalidStoredVersion(raw)
@@ -47,20 +59,15 @@ public final class UserVersionStorage<
return version
}
/// Stores the given version in the `user_version` field.
/// Stores the specified schema version in the `user_version` field.
///
/// This method updates the `PRAGMA user_version` field
/// with the raw `UInt32` value of the provided `Version`.
/// Updates the SQLite `PRAGMA user_version` value with the raw `UInt32` representation of the
/// provided ``Version``.
///
/// - Parameters:
/// - connection: The database connection.
/// - connection: The active database connection.
/// - version: The version to store.
public func setVersion(
_ connection: Connection,
_ version: Version
) throws {
connection.userVersion = .init(
bitPattern: version.rawValue
)
public func setVersion(_ connection: ConnectionProtocol, _ version: Version) throws {
connection.userVersion = .init(bitPattern: version.rawValue)
}
}