OK I have figured something out.
I created a DatabaseSnapshot that can store more than one model type and added the code necessary to decode the different types and extract properties from the snapshot.
No doubt there are more elegant ways of doing this but for now at least the basic custom DataStore is able to save and fetch records from PostgreSQL
```
//
// main.swift
// PGSQLDataStore
//
// Created by Duncan Groenewald on 1/9/2026.
//
import Foundation
import SwiftData
import PostgresNIO
import Logging
// ==========================================
// 1. DATA MODEL SCHEMA
// ==========================================
@Model
final class ProductRecord {
@Attribute(.unique) var id: UUID = UUID()
var sku: String = ""
var stockQuantity: Int = 0
init(id: UUID = UUID(), sku: String, stockQuantity: Int) {
self.id = id
self.sku = sku
self.stockQuantity = stockQuantity
}
enum CodingKeys: String, CodingKey {
case id, sku, stockQuantity
}
required init(from decoder: any Decoder) throws {
let container = try decoder.container(keyedBy: CodingKeys.self)
self.id = try container.decode(UUID.self, forKey: .id)
self.sku = try container.decode(String.self, forKey: .sku)
self.stockQuantity = try container.decode(Int.self, forKey: .stockQuantity)
}
func encode(to encoder: any Encoder) throws {
var container = encoder.container(keyedBy: CodingKeys.self)
try container.encode(id, forKey: .id)
try container.encode(sku, forKey: .sku)
try container.encode(stockQuantity, forKey: .stockQuantity)
}
}
@Model
final class SupplierRecord {
@Attribute(.unique) var id: UUID = UUID()
@Attribute(.unique) var name: String = ""
init(id: UUID = UUID(), name: String) {
self.id = id
self.name = name
}
enum CodingKeys: String, CodingKey {
case id, name
}
required init(from decoder: any Decoder) throws {
let container = try decoder.container(keyedBy: CodingKeys.self)
self.id = try container.decode(UUID.self, forKey: .id)
self.name = try container.decode(String.self, forKey: .name)
}
func encode(to encoder: any Encoder) throws {
var container = encoder.container(keyedBy: CodingKeys.self)
try container.encode(id, forKey: .id)
try container.encode(name, forKey: .name)
}
}
enum AppError: Error {
case dbError(String)
case userError(String)
case runtimeError(String)
case textFieldValidation(String)
}
extension AppError: LocalizedError {
public var errorDescription: String? {
switch self {
case let .textFieldValidation(message):
return NSLocalizedString(message, comment: "Validation error")
case let .dbError(message):
return NSLocalizedString(message, comment: "Data error")
case let .userError(message):
return NSLocalizedString(message, comment: "User error")
case let .runtimeError(message):
return NSLocalizedString(message, comment: "Runtime error")
}
}
}
enum PostgresSnapshotPayload: Codable {
case product(id: UUID, sku: String, stockQuantity: Int)
case supplier(id: UUID, name: String)
}
struct PostgresSnapshot: DataStoreSnapshot {
// Required protocol property identifier
var persistentIdentifier: PersistentIdentifier
// Poly-morphic storage payload
var payload: PostgresSnapshotPayload
// Requirement A: SwiftData extracts model state through this lifecycle initializer
init(from backingData: any BackingData, relatedBackingDatas: inout [PersistentIdentifier : any BackingData]) {
self.persistentIdentifier = backingData.persistentModelID!
let entityName = persistentIdentifier.entityName
switch entityName {
case "ProductRecord":
if let productBacking = backingData as? (any BackingData<ProductRecord>) {
let id = productBacking.getValue(forKey: \ProductRecord.id)
let sku = productBacking.getValue(forKey: \ProductRecord.sku)
let stock = productBacking.getValue(forKey: \ProductRecord.stockQuantity)
self.payload = .product(id: id, sku: sku, stockQuantity: stock)
} else {
self.payload = .product(id: UUID(), sku: "UNKNOWN", stockQuantity: 0)
}
case "SupplierRecord":
if let supplierBacking = backingData as? (any BackingData<SupplierRecord>) {
let id = supplierBacking.getValue(forKey: \SupplierRecord.id)
let name = supplierBacking.getValue(forKey: \SupplierRecord.name)
self.payload = .supplier(id: id, name: name)
} else {
self.payload = .supplier(id: UUID(), name: "UNKNOWN")
}
default:
fatalError("Unsupported entity schema variant: \(entityName)")
}
}
/// If we want to encode to some model type how do we know which type
func encode(to encoder: any Encoder) throws {
switch self.payload {
case .product(let id, let sku, let stockQuantity):
var container = encoder.container(keyedBy: ProductRecord.CodingKeys.self)
try container.encode(id, forKey: .id)
try container.encode(sku, forKey: .sku)
try container.encode(stockQuantity, forKey: .stockQuantity)
case .supplier(let id, let name):
var container = encoder.container(keyedBy: SupplierRecord.CodingKeys.self)
try container.encode(id, forKey: .id)
try container.encode(name, forKey: .name)
}
}
// Requirement B: Identifiers copier used during transaction insertions and cache remappings
func copy(persistentIdentifier: PersistentIdentifier, remappedIdentifiers: [PersistentIdentifier : PersistentIdentifier]? = nil) -> PostgresSnapshot {
return PostgresSnapshot(persistentIdentifier: persistentIdentifier, payload: self.payload)
}
// Custom initializers utilized for building records dynamically from Postgres rows
init(persistentIdentifier: PersistentIdentifier, payload: PostgresSnapshotPayload) {
self.persistentIdentifier = persistentIdentifier
self.payload = payload
}
}
//
// ==========================================
// 3. POSTGRESNIO STORES CONFIGURATION
// ==========================================
struct PostgresStoreConfiguration: DataStoreConfiguration {
typealias Store = PostgresSwiftDataStore
var name: String
var schema: Schema?
var connectionOptions: PostgresConnection.Configuration
}
extension PostgresStoreConfiguration: Equatable, Hashable {
static func == (lhs: PostgresStoreConfiguration, rhs: PostgresStoreConfiguration) -> Bool {
return lhs.name == rhs.name
}
func hash(into hasher: inout Hasher) {
hasher.combine(name)
}
}
// ==========================================
// 4. CUSTOM COMPLIANT DATASTORE ENGINE
// ==========================================
final class PostgresSwiftDataStore: DataStore {
let schema: Schema = Schema([ProductRecord.self, SupplierRecord.self], version: Schema.Version(1, 0, 0))
typealias Configuration = PostgresStoreConfiguration
typealias Snapshot = PostgresSnapshot
let configuration: PostgresStoreConfiguration
var connection: PostgresConnection? = nil
let logger = Logger(label: "com.app.postgres.datastore")
var identifier: String { "PostgresSwiftDataStore-\(configuration.name)" }
init(_ configuration: PostgresStoreConfiguration, migrationPlan: (any SchemaMigrationPlan.Type)?) throws {
self.configuration = configuration
let semaphore = DispatchSemaphore(value: 0)
Task {
do {
// Bootstrapping PostgresNIO wire framework connection
self.connection = try await PostgresConnection.connect(
on: MultiThreadedEventLoopGroup.singleton.any(),
configuration: configuration.connectionOptions,
id: 1,
logger: self.logger
)
} catch {
}
semaphore.signal()
}
semaphore.wait()
}
deinit {
self.close()
}
func close() {
let semaphore = DispatchSemaphore(value: 0)
Task {
try? await self.connection?.close()
semaphore.signal()
}
semaphore.wait()
}
// Saves changes by parsing individual items inside the save request bundle
func save(_ request: DataStoreSaveChangesRequest<PostgresSnapshot>) throws -> DataStoreSaveChangesResult<PostgresSnapshot> {
guard let connection = self.connection else {
throw AppError.runtimeError("No connection to database")
}
let semaphore = DispatchSemaphore(value: 0)
var remappedIdentifiers: [PersistentIdentifier: PersistentIdentifier] = [:]
var saveError: AppError? = nil
Task {
var remappedIdentifiersX: [PersistentIdentifier: PersistentIdentifier] = [:]
do {
// Process Insertions
for snapshot in request.inserted {
let entityName = snapshot.persistentIdentifier.entityName
switch snapshot.payload {
case .product(let id, let sku, let stockQuantity):
let pId = try PersistentIdentifier.identifier(for: identifier, entityName: entityName, primaryKey: id)
remappedIdentifiersX[snapshot.persistentIdentifier] = pId
try await connection.query(
"INSERT INTO products (id, sku, stockquantity) VALUES (\(id), \(sku), \(stockQuantity));",
logger: logger
)
case .supplier(let id, let name):
let pId = try PersistentIdentifier.identifier(for: identifier, entityName: entityName, primaryKey: id)
remappedIdentifiersX[snapshot.persistentIdentifier] = pId
try await connection.query(
"INSERT INTO suppliers (id, name) VALUES (\(id), \(name));",
logger: logger
)
}
}
// Process Modifications (Updates)
for snapshot in request.updated {
switch snapshot.payload {
case .product(let id, let sku, let stockQuantity):
try await connection.query(
"UPDATE products SET sku = \(sku), stockquantity = \(stockQuantity) WHERE id = \(id);",
logger: logger
)
case .supplier(let id, let name):
try await connection.query(
"UPDATE suppliers SET name = \(name) WHERE id = \(id);",
logger: logger
)
}
}
// Process Record Deletions
for snapshot in request.deleted {
switch snapshot.payload {
case .product(let id, _, _):
try await connection.query("DELETE FROM products WHERE id = \(id);", logger: logger)
case .supplier(let id, _):
try await connection.query("DELETE FROM suppliers WHERE id = \(id);", logger: logger)
}
}
} catch {
logger.error("Failed executing batch change block: \(error)")
saveError = AppError.runtimeError("Failed executing batch change block: \(String(reflecting:error))")
}
remappedIdentifiers = remappedIdentifiersX
semaphore.signal()
}
semaphore.wait()
if saveError != nil {
throw saveError!
}
return DataStoreSaveChangesResult(for: self.identifier, remappedIdentifiers: remappedIdentifiers)
}
// Reads out records and hydrates fresh snapshot frames directly
func fetch<T>(_ request: DataStoreFetchRequest<T>) throws -> DataStoreFetchResult<T, PostgresSnapshot> where T : PersistentModel {
guard let connection = self.connection else {
throw AppError.runtimeError("No connection to database")
}
var collectedSnapshots: [PostgresSnapshot] = []
let semaphore = DispatchSemaphore(value: 0)
// Target validation checks based on the incoming context model type requested
let modelName = String(describing: T.self)
Task {
do {
if modelName == "ProductRecord" {
let rows = try await connection.query(
"SELECT id, sku, stockquantity FROM products;",
logger: logger
)
// Decode rows directly into tuples via PostgresNIO
for try await (dbID, dbSku, dbStock) in rows.decode((UUID, String, Int).self) {
let persistentID = try PersistentIdentifier.identifier(for: identifier, entityName: modelName, primaryKey: dbID)
let snapshot = PostgresSnapshot(
persistentIdentifier: persistentID,
payload: .product(id: dbID, sku: dbSku, stockQuantity: dbStock)
)
collectedSnapshots.append(snapshot)
}
} else if modelName == "SupplierRecord" {
let rows = try await connection.query(
"SELECT id, name FROM suppliers;",
logger: logger
)
// Decode rows directly into tuples via PostgresNIO
for try await (dbID, dbName) in rows.decode((UUID, String).self) {
let persistentID = try PersistentIdentifier.identifier(for: identifier, entityName: modelName, primaryKey: dbID)
let snapshot = PostgresSnapshot(
persistentIdentifier: persistentID,
payload: .supplier(id: dbID, name: dbName)
)
collectedSnapshots.append(snapshot)
}
}
} catch {
logger.error("Data tracking stream operation caught error: \(error)")
}
semaphore.signal()
}
semaphore.wait()
return DataStoreFetchResult(descriptor: request.descriptor, fetchedSnapshots: collectedSnapshots)
}
}
struct Credentials {
var username: String
var password: String
}
// ==========================================
// 5. RUNTIME INITIALIZER
// ==========================================
struct InventoryApp {
static let hostname: String = "localhost"
static let username: String = "swiftdatastore"
static let databasename: String = "swiftdatastore"
static let port: Int = 5432
static let password: String = "swiftdatastore"
static func main() async {
print("🚀 Initializing Protocol-Compliant PostgresNIO + SwiftData Core Context...")
let credentials = Credentials(username: username, password: password)
var tlsConfiguration = TLSConfiguration.makeClientConfiguration()
tlsConfiguration.certificateVerification = .none
// let config = PostgresClient.Configuration(
// host: hostname,
// port: port,
// username: username,
// password: credentials.password,
// database: databasename,
// tls: .prefer(tlsConfiguration)
// )
//
let storeConfig = PostgresStoreConfiguration(
name: "ProductionInventory",
schema: Schema([SupplierRecord.self, ProductRecord.self]),
connectionOptions: try! createPgConConfiguration()
)
var context : ModelContext? = nil
do {
// Instantiate our updated custom store natively into the model container
let customContainer = try ModelContainer(for: SupplierRecord.self, ProductRecord.self, configurations: storeConfig)
context = ModelContext(customContainer)
} catch {
print("🛑 Fatal exception creating model container: \(String(reflecting:error.localizedDescription))")
return
}
guard let context = context else {
print("Error opening model context, unable to continue!")
return
}
do {
// --- INSERTS TESTING ---
print("💾 Saving a new record into SwiftData context...")
let newSupplier = SupplierRecord(name: "Supplier 1")
context.insert(newSupplier)
try context.save()
let newSupplier2 = SupplierRecord(name: "Supplier 2")
context.insert(newSupplier2)
try context.save()
let newProduct = ProductRecord(sku: "IPHONE-16-PRO", stockQuantity: 20)
context.insert(newProduct)
try context.save()
let newProduct2 = ProductRecord(sku: "IPHONE-15-PRO", stockQuantity: 50)
context.insert(newProduct2)
try context.save()
} catch {
print("🛑 Fatal exception creating records: \(String(reflecting:error.localizedDescription))")
}
do {
// --- FETCHES TESTING ---
print("🔍 Requesting collection from PostgreSQL target tables...")
let fetchDescriptor1 = FetchDescriptor<SupplierRecord>()
let suppliers = try context.fetch(fetchDescriptor1)
let fetchDescriptor2 = FetchDescriptor<ProductRecord>()
let products = try context.fetch(fetchDescriptor2)
for item in suppliers {
print("📦 Mapped Record -> Identifier: \(item.id.uuidString) | Name: \(item.name) ")
}
for item in products {
//print("📦 Mapped Record -> SKU: \(item.sku) | Stock: \(item.stockQuantity) | Identifier: \(item.id.uuidString)")
print("📦 Mapped Record -> Identifier: \(item.id.uuidString) | SKU: \(item.sku) | Stock: \(item.stockQuantity)")
}
} catch {
print("🛑 Fatal exception fetching records: \(String(reflecting:error.localizedDescription))")
}
}
static func createPgConConfiguration() throws -> PostgresConnection.Configuration {
let credentials = Credentials(username: username, password: password)
var tlsConfiguration = TLSConfiguration.makeClientConfiguration()
tlsConfiguration.certificateVerification = .none
let sslContext = try! NIOSSLContext(configuration: tlsConfiguration)
//var tlsConfiguration = TLSConfiguration.makeClientConfiguration()
let config = PostgresConnection.Configuration(
host: hostname,
port: port,
username: username,
password: credentials.password,
database: databasename,
tls: .prefer(sslContext)
)
return config
}
}
await InventoryApp.main()
/* DATABASE SCHEMA */
/*
CREATE TABLE swiftdatastore.products (
id UUID PRIMARY KEY NOT NULL DEFAULT (gen_random_uuid()),
tstamp timestamp with time zone DeFAULT CURRENT_TIMESTAMP,
sku text NOT NULL,
stockQuantity int DEFAULT 0,
UNIQUE (sku)
);
CREATE TABLE swiftdatastore.suppliers (
id UUID PRIMARY KEY NOT NULL DEFAULT (gen_random_uuid()),
tstamp timestamp with time zone DeFAULT CURRENT_TIMESTAMP,
name text NOT NULL,
UNIQUE (name)
);
*/
// ==========================================
//// 2. CUSTOM SNAPSHOT
//// ==========================================
//struct ProductPostgresSnapshot: DataStoreSnapshot {
// // Required protocol property identifier
// var persistentIdentifier: PersistentIdentifier
//
// // Strict column fields matching PostgreSQL definitions
// var id: UUID
// var sku: String
// var stockQuantity: Int
//
// // Requirement A: Lifecycle initializer used by SwiftData to pull out internal model states
// init(from backingData: any BackingData, relatedBackingDatas: inout [PersistentIdentifier : any BackingData]) {
// // Recover or safely establish the identity block
// self.persistentIdentifier = backingData.persistentModelID!
//
// // Dynamically extract model values from the type-erased BackingData container
// if let userBacking = backingData as? (any BackingData<ProductRecord>) {
// self.id = userBacking.getValue(forKey: \ProductRecord.id)
// self.sku = userBacking.getValue(forKey: \ProductRecord.sku)
// self.stockQuantity = userBacking.getValue(forKey: \ProductRecord.stockQuantity)
// } else {
// // Safe fallbacks for edge-case corruption payloads
// self.id = UUID()
// self.sku = "UNKNOWN"
// self.stockQuantity = 0
// }
// }
//
// // Requirement B: Identifiers copier used during transaction insertions and cache remappings
// func copy(persistentIdentifier: PersistentIdentifier, remappedIdentifiers: [PersistentIdentifier : PersistentIdentifier]? = nil) -> ProductPostgresSnapshot {
// return ProductPostgresSnapshot(
// persistentIdentifier: persistentIdentifier,
// id: self.id,
// sku: self.sku,
// stockQuantity: self.stockQuantity
// )
// }
// enum CodingKeys: String, CodingKey {
// case persistentIdentifier, id, sku, stockQuantity
// }
// init(from decoder: any Decoder) throws {
// let container = try decoder.container(keyedBy: CodingKeys.self)
// self.persistentIdentifier = try container.decode(PersistentIdentifier.self, forKey: .persistentIdentifier)
// self.id = try container.decode(UUID.self, forKey: .id)
// self.sku = try container.decode(String.self, forKey: .sku)
// self.stockQuantity = try container.decode(Int.self, forKey: .stockQuantity)
// }
//
// func encode(to encoder: any Encoder) throws {
// var container = encoder.container(keyedBy: CodingKeys.self)
// try container.encode(persistentIdentifier, forKey: .persistentIdentifier)
// try container.encode(id, forKey: .id)
// try container.encode(sku, forKey: .sku)
// try container.encode(stockQuantity, forKey: .stockQuantity)
// }
//
// // Custom driver initializer utilized for building records dynamically from PostgresNIO rows
// init(persistentIdentifier: PersistentIdentifier, id: UUID, sku: String, stockQuantity: Int) {
// self.persistentIdentifier = persistentIdentifier
// self.id = id
// self.sku = sku
// self.stockQuantity = stockQuantity
// }
//}
```
I seem to have found a solution which seems to be that the encoding and decoding functions needed to be defined.
The code below now seems to save and fetch records from PostgreSQL database.
```
//
// main.swift
// PGSQLDataStore
//
// Created by Duncan Groenewald on 1/9/2026.
//
import Foundation
import SwiftData
import PostgresNIO
import Logging
// ==========================================
// 1. DATA MODEL SCHEMA
// ==========================================
@Model
final class ProductRecord {
@Attribute(.unique) var id: UUID = UUID()
var sku: String = ""
var stockQuantity: Int = 0
init(id: UUID = UUID(), sku: String, stockQuantity: Int) {
self.id = id
self.sku = sku
self.stockQuantity = stockQuantity
}
enum CodingKeys: String, CodingKey {
case id, sku, stockQuantity
}
required init(from decoder: any Decoder) throws {
let container = try decoder.container(keyedBy: CodingKeys.self)
self.id = try container.decode(UUID.self, forKey: .id)
self.sku = try container.decode(String.self, forKey: .sku)
self.stockQuantity = try container.decode(Int.self, forKey: .stockQuantity)
}
func encode(to encoder: any Encoder) throws {
var container = encoder.container(keyedBy: CodingKeys.self)
try container.encode(id, forKey: .id)
try container.encode(sku, forKey: .sku)
try container.encode(stockQuantity, forKey: .stockQuantity)
}
}
@Model
final class SupplierRecord {
@Attribute(.unique) var id: UUID = UUID()
@Attribute(.unique) var name: String = ""
init(id: UUID = UUID(), name: String) {
self.id = id
self.name = name
}
}
enum AppError: Error {
case runtimeError(String)
}
// ==========================================
// 2. CUSTOM SNAPSHOT
// ==========================================
struct ProductPostgresSnapshot: DataStoreSnapshot {
// Required protocol property identifier
var persistentIdentifier: PersistentIdentifier
// Strict column fields matching PostgreSQL definitions
var id: UUID
var sku: String
var stockQuantity: Int
// Requirement A: Lifecycle initializer used by SwiftData to pull out internal model states
init(from backingData: any BackingData, relatedBackingDatas: inout [PersistentIdentifier : any BackingData]) {
// Recover or safely establish the identity block
self.persistentIdentifier = backingData.persistentModelID!
// Dynamically extract model values from the type-erased BackingData container
if let userBacking = backingData as? (any BackingData<ProductRecord>) {
self.id = userBacking.getValue(forKey: \ProductRecord.id)
self.sku = userBacking.getValue(forKey: \ProductRecord.sku)
self.stockQuantity = userBacking.getValue(forKey: \ProductRecord.stockQuantity)
} else {
// Safe fallbacks for edge-case corruption payloads
self.id = UUID()
self.sku = "UNKNOWN"
self.stockQuantity = 0
}
}
// Requirement B: Identifiers copier used during transaction insertions and cache remappings
func copy(persistentIdentifier: PersistentIdentifier, remappedIdentifiers: [PersistentIdentifier : PersistentIdentifier]? = nil) -> ProductPostgresSnapshot {
return ProductPostgresSnapshot(
persistentIdentifier: persistentIdentifier,
id: self.id,
sku: self.sku,
stockQuantity: self.stockQuantity
)
}
enum CodingKeys: String, CodingKey {
case persistentIdentifier, id, sku, stockQuantity
}
init(from decoder: any Decoder) throws {
let container = try decoder.container(keyedBy: CodingKeys.self)
self.persistentIdentifier = try container.decode(PersistentIdentifier.self, forKey: .persistentIdentifier)
self.id = try container.decode(UUID.self, forKey: .id)
self.sku = try container.decode(String.self, forKey: .sku)
self.stockQuantity = try container.decode(Int.self, forKey: .stockQuantity)
}
func encode(to encoder: any Encoder) throws {
var container = encoder.container(keyedBy: CodingKeys.self)
try container.encode(persistentIdentifier, forKey: .persistentIdentifier)
try container.encode(id, forKey: .id)
try container.encode(sku, forKey: .sku)
try container.encode(stockQuantity, forKey: .stockQuantity)
}
// Custom driver initializer utilized for building records dynamically from PostgresNIO rows
init(persistentIdentifier: PersistentIdentifier, id: UUID, sku: String, stockQuantity: Int) {
self.persistentIdentifier = persistentIdentifier
self.id = id
self.sku = sku
self.stockQuantity = stockQuantity
}
}
// ==========================================
// 3. POSTGRESNIO STORES CONFIGURATION
// ==========================================
struct PostgresStoreConfiguration: DataStoreConfiguration {
typealias Store = PostgresSwiftDataStore
var name: String
var schema: Schema?
var connectionOptions: PostgresConnection.Configuration
}
extension PostgresStoreConfiguration: Equatable, Hashable {
static func == (lhs: PostgresStoreConfiguration, rhs: PostgresStoreConfiguration) -> Bool {
return lhs.name == rhs.name
}
func hash(into hasher: inout Hasher) {
hasher.combine(name)
}
}
// ==========================================
// 4. CUSTOM COMPLIANT DATASTORE ENGINE
// ==========================================
final class PostgresSwiftDataStore: DataStore {
let schema: Schema = Schema([ProductRecord.self], version: Schema.Version(1, 0, 0))
typealias Configuration = PostgresStoreConfiguration
typealias Snapshot = ProductPostgresSnapshot
let configuration: PostgresStoreConfiguration
var connection: PostgresConnection? = nil
let logger = Logger(label: "com.app.postgres.datastore")
var identifier: String { "PostgresSwiftDataStore-\(configuration.name)" }
init(_ configuration: PostgresStoreConfiguration, migrationPlan: (any SchemaMigrationPlan.Type)?) throws {
self.configuration = configuration
let semaphore = DispatchSemaphore(value: 0)
Task {
do {
// Bootstrapping PostgresNIO wire framework connection
self.connection = try await PostgresConnection.connect(
on: MultiThreadedEventLoopGroup.singleton.any(),
configuration: configuration.connectionOptions,
id: 1,
logger: self.logger
)
} catch {
}
semaphore.signal()
}
semaphore.wait()
}
deinit {
self.close()
}
func close() {
let semaphore = DispatchSemaphore(value: 0)
Task {
try? await self.connection?.close()
semaphore.signal()
}
semaphore.wait()
}
// Saves changes by parsing individual items inside the save request bundle
func save(_ request: DataStoreSaveChangesRequest<ProductPostgresSnapshot>) throws -> DataStoreSaveChangesResult<ProductPostgresSnapshot> {
guard let connection = self.connection else {
throw AppError.runtimeError("No connection to database")
}
let semaphore = DispatchSemaphore(value: 0)
var remappedIdentifiers: [PersistentIdentifier: PersistentIdentifier] = [:]
Task {
var remappedIdentifiersX: [PersistentIdentifier: PersistentIdentifier] = [:]
do {
// Process Insertions
for snapshot in request.inserted {
print("\(snapshot.persistentIdentifier.storeIdentifier ?? "nil"), \(snapshot.persistentIdentifier.entityName), \(snapshot.persistentIdentifier.id)")
let pId = try PersistentIdentifier.identifier(for: identifier, entityName: "ProductRecord", primaryKey: snapshot.id)
remappedIdentifiersX[snapshot.persistentIdentifier] = pId
try await connection.query(
"""
INSERT INTO products (id, sku, stockquantity)
VALUES (\(snapshot.id), \(snapshot.sku), \(snapshot.stockQuantity));
""",
logger: logger
)
}
// Process Modifications (Updates)
for snapshot in request.updated {
try await connection.query(
"""
UPDATE items_inventory
SET sku = \(snapshot.sku), stockquantity = \(snapshot.stockQuantity)
WHERE id = \(snapshot.id);
""",
logger: logger
)
}
// Process Record Deletions
for snapshot in request.deleted {
try await connection.query(
"DELETE FROM products WHERE id = \(snapshot.id);",
logger: logger
)
}
} catch {
logger.error("Failed executing batch change block: \(error)")
}
remappedIdentifiers = remappedIdentifiersX
semaphore.signal()
}
semaphore.wait()
return DataStoreSaveChangesResult(for: self.identifier, remappedIdentifiers: remappedIdentifiers)
}
// Reads out records and hydrates fresh snapshot frames directly
func fetch<T>(_ request: DataStoreFetchRequest<T>) throws -> DataStoreFetchResult<T, ProductPostgresSnapshot> where T : PersistentModel {
guard let connection = self.connection else {
throw AppError.runtimeError("No connection to database")
}
var collectedSnapshots: [ProductPostgresSnapshot] = []
let semaphore = DispatchSemaphore(value: 0)
Task {
do {
let rows = try await connection.query(
"SELECT id, sku, stockquantity FROM products;",
logger: logger
)
// Decode rows directly into tuples via PostgresNIO
for try await (dbID, dbSku, dbStock) in rows.decode((UUID, String, Int).self) {
let persistentID = try PersistentIdentifier.identifier(for: identifier, entityName: "ProductRecord", primaryKey: dbID)
let snapshot = ProductPostgresSnapshot(
persistentIdentifier: persistentID,
id: dbID,
sku: dbSku,
stockQuantity: dbStock
)
collectedSnapshots.append(snapshot)
}
} catch {
logger.error("Data tracking stream operation caught error: \(error)")
}
semaphore.signal()
}
semaphore.wait()
return DataStoreFetchResult(descriptor: request.descriptor, fetchedSnapshots: collectedSnapshots)
}
}
struct Credentials {
var username: String
var password: String
}
// ==========================================
// 5. RUNTIME INITIALIZER
// ==========================================
struct InventoryApp {
static let hostname: String = "localhost"
static let username: String = "swiftdatastore"
static let databasename: String = "swiftdatastore"
static let port: Int = 5432
static let password: String = "swiftdatastore"
static func main() async {
print("🚀 Initializing Protocol-Compliant PostgresNIO + SwiftData Core Context...")
let credentials = Credentials(username: username, password: password)
var tlsConfiguration = TLSConfiguration.makeClientConfiguration()
tlsConfiguration.certificateVerification = .none
let config = PostgresClient.Configuration(
host: hostname,
port: port,
username: username,
password: credentials.password,
database: databasename,
tls: .prefer(tlsConfiguration)
)
let storeConfig = PostgresStoreConfiguration(
name: "ProductionInventory",
schema: Schema([ProductRecord.self]),
connectionOptions: try! createPgConConfiguration()
)
do {
// Instantiate our updated custom store natively into the model container
let customContainer = try ModelContainer(for: ProductRecord.self, configurations: storeConfig)
let context = ModelContext(customContainer)
// --- INSERTS TESTING ---
print("💾 Saving a new record into SwiftData context...")
let newProduct = ProductRecord(sku: "IPHONE-18-PRO", stockQuantity: 150)
context.insert(newProduct)
try context.save()
let newProduct2 = ProductRecord(sku: "IPHONE-17-PRO", stockQuantity: 100)
context.insert(newProduct2)
try context.save()
// --- FETCHES TESTING ---
print("🔍 Requesting collection from PostgreSQL target tables...")
let fetchDescriptor = FetchDescriptor<ProductRecord>()
let localRecords = try context.fetch(fetchDescriptor)
for item in localRecords {
//print("📦 Mapped Record -> SKU: \(item.sku) | Stock: \(item.stockQuantity) | Identifier: \(item.id.uuidString)")
print("📦 Mapped Record -> Identifier: \(item.id.uuidString) | SKU: \(item.sku) | Stock: \(item.stockQuantity)")
}
} catch {
print("🛑 Fatal exception caught in pipeline execution: \(error)")
}
}
static func createPgConConfiguration() throws -> PostgresConnection.Configuration {
let credentials = Credentials(username: username, password: password)
var tlsConfiguration = TLSConfiguration.makeClientConfiguration()
tlsConfiguration.certificateVerification = .none
let sslContext = try! NIOSSLContext(configuration: tlsConfiguration)
//var tlsConfiguration = TLSConfiguration.makeClientConfiguration()
let config = PostgresConnection.Configuration(
host: hostname,
port: port,
username: username,
password: credentials.password,
database: databasename,
tls: .prefer(sslContext)
)
return config
}
}
await InventoryApp.main()
/* DATABASE SCHEMA */
/*
CREATE TABLE swiftdatastore.products (
id UUID PRIMARY KEY NOT NULL DEFAULT (gen_random_uuid()),
tstamp timestamp with time zone DeFAULT CURRENT_TIMESTAMP,
sku text NOT NULL,
stockQuantity int DEFAULT 0,
UNIQUE (sku)
);
*/
```
