|
|
|
|
@ -10,6 +10,12 @@ import DifferenceKit
|
|
|
|
|
///
|
|
|
|
|
/// **Note:** We **MUST** have accurate `filterSQL` and `orderSQL` values otherwise the indexing won't work
|
|
|
|
|
public class PagedDatabaseObserver<ObservedTable, T>: TransactionObserver where ObservedTable: TableRecord & ColumnExpressible & Identifiable, T: FetchableRecordWithRowId & Identifiable {
|
|
|
|
|
private let commitProcessingQueue: DispatchQueue = DispatchQueue(
|
|
|
|
|
label: "PagedDatabaseObserver.commitProcessingQueue",
|
|
|
|
|
qos: .userInitiated,
|
|
|
|
|
attributes: [] // Must be serial in order to avoid updates getting processed in the wrong order
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
// MARK: - Variables
|
|
|
|
|
|
|
|
|
|
private let pagedTableName: String
|
|
|
|
|
@ -145,74 +151,58 @@ public class PagedDatabaseObserver<ObservedTable, T>: TransactionObserver where
|
|
|
|
|
changesInCommit.mutate { $0.insert(trackedChange) }
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// Note: We will process all updates which come through this method even if
|
|
|
|
|
// 'onChange' is null because if the UI stops observing and then starts again
|
|
|
|
|
// later we don't want to have missed any changes which happened while the UI
|
|
|
|
|
// wasn't subscribed (and doing a full re-query seems painful...)
|
|
|
|
|
/// We will process all updates which come through this method even if 'onChange' is null because if the UI stops observing and then starts
|
|
|
|
|
/// again later we don't want to have missed any changes which happened while the UI wasn't subscribed (and doing a full re-query seems painful...)
|
|
|
|
|
///
|
|
|
|
|
/// **Note:** This function is generally called within the DBWrite thread but we don't actually need write access to process the commit, in order
|
|
|
|
|
/// to avoid blocking the DBWrite thread we dispatch to a serial `commitProcessingQueue` to process the incoming changes (in the past not doing
|
|
|
|
|
/// so was resulting in hanging when there was a lot of activity happening)
|
|
|
|
|
public func databaseDidCommit(_ db: Database) {
|
|
|
|
|
// If there were no pending changes in the commit then do nothing
|
|
|
|
|
guard !self.changesInCommit.wrappedValue.isEmpty else { return }
|
|
|
|
|
|
|
|
|
|
// Since we can't be sure the behaviours of 'databaseDidChange' and 'databaseDidCommit' won't change in
|
|
|
|
|
// the future we extract and clear the values in 'changesInCommit' since it's 'Atomic<T>' so will different
|
|
|
|
|
// threads modifying the data resulting in us missing a change
|
|
|
|
|
var committedChanges: Set<PagedData.TrackedChange> = []
|
|
|
|
|
|
|
|
|
|
self.changesInCommit.mutate { cachedChanges in
|
|
|
|
|
committedChanges = cachedChanges
|
|
|
|
|
cachedChanges.removeAll()
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// Note: This method will be called regardless of whether there were actually changes
|
|
|
|
|
// in the areas we are observing so we want to early-out if there aren't any relevant
|
|
|
|
|
// updated rows
|
|
|
|
|
guard !committedChanges.isEmpty else { return }
|
|
|
|
|
commitProcessingQueue.async { [weak self] in
|
|
|
|
|
self?.processDatabaseCommit(committedChanges: committedChanges)
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
private func processDatabaseCommit(committedChanges: Set<PagedData.TrackedChange>) {
|
|
|
|
|
typealias AssociatedDataInfo = [(hasChanges: Bool, data: ErasedAssociatedRecord)]
|
|
|
|
|
typealias UpdatedData = (cache: DataCache<T>, pageInfo: PagedData.PageInfo, hasChanges: Bool, associatedData: AssociatedDataInfo)
|
|
|
|
|
|
|
|
|
|
// Store the instance variables locally to avoid unwrapping
|
|
|
|
|
let dataCache: DataCache<T> = self.dataCache.wrappedValue
|
|
|
|
|
let pageInfo: PagedData.PageInfo = self.pageInfo.wrappedValue
|
|
|
|
|
let joinSQL: SQL? = self.joinSQL
|
|
|
|
|
let orderSQL: SQL = self.orderSQL
|
|
|
|
|
let filterSQL: SQL = self.filterSQL
|
|
|
|
|
let associatedRecords: [ErasedAssociatedRecord] = self.associatedRecords
|
|
|
|
|
|
|
|
|
|
let updateDataAndCallbackIfNeeded: (DataCache<T>, PagedData.PageInfo, Bool) -> () = { [weak self] updatedDataCache, updatedPageInfo, cacheHasChanges in
|
|
|
|
|
let associatedDataInfo: [(hasChanges: Bool, data: ErasedAssociatedRecord)] = associatedRecords
|
|
|
|
|
.map { associatedRecord in
|
|
|
|
|
let hasChanges: Bool = associatedRecord.tryUpdateForDatabaseCommit(
|
|
|
|
|
db,
|
|
|
|
|
changes: committedChanges,
|
|
|
|
|
joinSQL: joinSQL,
|
|
|
|
|
orderSQL: orderSQL,
|
|
|
|
|
filterSQL: filterSQL,
|
|
|
|
|
pageInfo: updatedPageInfo
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
return (hasChanges, associatedRecord)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// Check if we need to trigger a change callback
|
|
|
|
|
guard cacheHasChanges || associatedDataInfo.contains(where: { hasChanges, _ in hasChanges }) else {
|
|
|
|
|
return
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// If the associated data changed then update the updatedCachedData with the
|
|
|
|
|
// updated associated data
|
|
|
|
|
var finalUpdatedDataCache: DataCache<T> = updatedDataCache
|
|
|
|
|
|
|
|
|
|
associatedDataInfo.forEach { hasChanges, associatedData in
|
|
|
|
|
guard cacheHasChanges || hasChanges else { return }
|
|
|
|
|
let getAssociatedDataInfo: (Database, PagedData.PageInfo) -> AssociatedDataInfo = { db, updatedPageInfo in
|
|
|
|
|
associatedRecords.map { associatedRecord in
|
|
|
|
|
let hasChanges: Bool = associatedRecord.tryUpdateForDatabaseCommit(
|
|
|
|
|
db,
|
|
|
|
|
changes: committedChanges,
|
|
|
|
|
joinSQL: joinSQL,
|
|
|
|
|
orderSQL: orderSQL,
|
|
|
|
|
filterSQL: filterSQL,
|
|
|
|
|
pageInfo: updatedPageInfo
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
finalUpdatedDataCache = associatedData.updateAssociatedData(to: finalUpdatedDataCache)
|
|
|
|
|
return (hasChanges, associatedRecord)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// Update the cache, pageInfo and the change callback
|
|
|
|
|
self?.dataCache.mutate { $0 = finalUpdatedDataCache }
|
|
|
|
|
self?.pageInfo.mutate { $0 = updatedPageInfo }
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
// Make sure the updates run on the main thread
|
|
|
|
|
guard Thread.isMainThread else {
|
|
|
|
|
DispatchQueue.main.async { [weak self] in
|
|
|
|
|
self?.onChangeUnsorted(finalUpdatedDataCache.values, updatedPageInfo)
|
|
|
|
|
}
|
|
|
|
|
return
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
self?.onChangeUnsorted(finalUpdatedDataCache.values, updatedPageInfo)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// Determing if there were any direct or related data changes
|
|
|
|
|
// Determine if there were any direct or related data changes
|
|
|
|
|
let directChanges: Set<PagedData.TrackedChange> = committedChanges
|
|
|
|
|
.filter { $0.tableName == pagedTableName }
|
|
|
|
|
let relatedChanges: [String: [PagedData.TrackedChange]] = committedChanges
|
|
|
|
|
@ -227,215 +217,248 @@ public class PagedDatabaseObserver<ObservedTable, T>: TransactionObserver where
|
|
|
|
|
.filter { $0.tableName != pagedTableName }
|
|
|
|
|
.filter { $0.kind == .delete }
|
|
|
|
|
|
|
|
|
|
guard !directChanges.isEmpty || !relatedChanges.isEmpty || !relatedDeletions.isEmpty else {
|
|
|
|
|
updateDataAndCallbackIfNeeded(self.dataCache.wrappedValue, self.pageInfo.wrappedValue, false)
|
|
|
|
|
return
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
var updatedPageInfo: PagedData.PageInfo = self.pageInfo.wrappedValue
|
|
|
|
|
var updatedDataCache: DataCache<T> = self.dataCache.wrappedValue
|
|
|
|
|
let deletionChanges: [Int64] = directChanges
|
|
|
|
|
.filter { $0.kind == .delete }
|
|
|
|
|
.map { $0.rowId }
|
|
|
|
|
let oldDataCount: Int = dataCache.wrappedValue.count
|
|
|
|
|
|
|
|
|
|
// First remove any items which have been deleted
|
|
|
|
|
if !deletionChanges.isEmpty {
|
|
|
|
|
updatedDataCache = updatedDataCache.deleting(rowIds: deletionChanges)
|
|
|
|
|
|
|
|
|
|
// Make sure there were actually changes
|
|
|
|
|
if updatedDataCache.count != oldDataCount {
|
|
|
|
|
let dataSizeDiff: Int = (updatedDataCache.count - oldDataCount)
|
|
|
|
|
// Process and retrieve the updated data
|
|
|
|
|
let updatedData: UpdatedData = Storage.shared
|
|
|
|
|
.read { db -> UpdatedData in
|
|
|
|
|
// If there aren't any direct or related changes then early-out
|
|
|
|
|
guard !directChanges.isEmpty || !relatedChanges.isEmpty || !relatedDeletions.isEmpty else {
|
|
|
|
|
return (dataCache, pageInfo, false, getAssociatedDataInfo(db, pageInfo))
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
updatedPageInfo = PagedData.PageInfo(
|
|
|
|
|
pageSize: updatedPageInfo.pageSize,
|
|
|
|
|
pageOffset: updatedPageInfo.pageOffset,
|
|
|
|
|
currentCount: (updatedPageInfo.currentCount + dataSizeDiff),
|
|
|
|
|
totalCount: (updatedPageInfo.totalCount + dataSizeDiff)
|
|
|
|
|
)
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// If there are no inserted/updated rows then trigger the update callback and stop here
|
|
|
|
|
let changesToQuery: [PagedData.TrackedChange] = directChanges
|
|
|
|
|
.filter { $0.kind != .delete }
|
|
|
|
|
|
|
|
|
|
guard !changesToQuery.isEmpty || !relatedChanges.isEmpty || !relatedDeletions.isEmpty else {
|
|
|
|
|
updateDataAndCallbackIfNeeded(updatedDataCache, updatedPageInfo, !deletionChanges.isEmpty)
|
|
|
|
|
return
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// First we need to get the rowIds for the paged data connected to any of the related changes
|
|
|
|
|
let pagedRowIdsForRelatedChanges: Set<Int64> = {
|
|
|
|
|
guard !relatedChanges.isEmpty else { return [] }
|
|
|
|
|
|
|
|
|
|
return relatedChanges
|
|
|
|
|
.reduce(into: []) { result, next in
|
|
|
|
|
guard
|
|
|
|
|
let observedChange: PagedData.ObservedChanges = observedTableChangeTypes[next.key],
|
|
|
|
|
let joinToPagedType: SQL = observedChange.joinToPagedType
|
|
|
|
|
else { return }
|
|
|
|
|
// Store a mutable copies of the dataCache and pageInfo for updating
|
|
|
|
|
var updatedDataCache: DataCache<T> = dataCache
|
|
|
|
|
var updatedPageInfo: PagedData.PageInfo = pageInfo
|
|
|
|
|
let deletionChanges: [Int64] = directChanges
|
|
|
|
|
.filter { $0.kind == .delete }
|
|
|
|
|
.map { $0.rowId }
|
|
|
|
|
let oldDataCount: Int = dataCache.count
|
|
|
|
|
|
|
|
|
|
// First remove any items which have been deleted
|
|
|
|
|
if !deletionChanges.isEmpty {
|
|
|
|
|
updatedDataCache = updatedDataCache.deleting(rowIds: deletionChanges)
|
|
|
|
|
|
|
|
|
|
let pagedRowIds: [Int64] = PagedData.pagedRowIdsForRelatedRowIds(
|
|
|
|
|
db,
|
|
|
|
|
tableName: next.key,
|
|
|
|
|
pagedTableName: pagedTableName,
|
|
|
|
|
relatedRowIds: Array(next.value.map { $0.rowId }.asSet()),
|
|
|
|
|
joinToPagedType: joinToPagedType
|
|
|
|
|
)
|
|
|
|
|
// Make sure there were actually changes
|
|
|
|
|
if updatedDataCache.count != oldDataCount {
|
|
|
|
|
let dataSizeDiff: Int = (updatedDataCache.count - oldDataCount)
|
|
|
|
|
|
|
|
|
|
updatedPageInfo = PagedData.PageInfo(
|
|
|
|
|
pageSize: updatedPageInfo.pageSize,
|
|
|
|
|
pageOffset: updatedPageInfo.pageOffset,
|
|
|
|
|
currentCount: (updatedPageInfo.currentCount + dataSizeDiff),
|
|
|
|
|
totalCount: (updatedPageInfo.totalCount + dataSizeDiff)
|
|
|
|
|
)
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// If there are no inserted/updated rows then trigger then early-out
|
|
|
|
|
let changesToQuery: [PagedData.TrackedChange] = directChanges
|
|
|
|
|
.filter { $0.kind != .delete }
|
|
|
|
|
|
|
|
|
|
guard !changesToQuery.isEmpty || !relatedChanges.isEmpty || !relatedDeletions.isEmpty else {
|
|
|
|
|
let associatedData: AssociatedDataInfo = getAssociatedDataInfo(db, updatedPageInfo)
|
|
|
|
|
return (updatedDataCache, updatedPageInfo, !deletionChanges.isEmpty, associatedData)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// Next we need to determine if any related changes were associated to the pagedData we are
|
|
|
|
|
// observing, if they aren't (and there were no other direct changes) we can early-out
|
|
|
|
|
let pagedRowIdsForRelatedChanges: Set<Int64> = {
|
|
|
|
|
guard !relatedChanges.isEmpty else { return [] }
|
|
|
|
|
|
|
|
|
|
result.append(contentsOf: pagedRowIds)
|
|
|
|
|
return relatedChanges
|
|
|
|
|
.reduce(into: []) { result, next in
|
|
|
|
|
guard
|
|
|
|
|
let observedChange: PagedData.ObservedChanges = observedTableChangeTypes[next.key],
|
|
|
|
|
let joinToPagedType: SQL = observedChange.joinToPagedType
|
|
|
|
|
else { return }
|
|
|
|
|
|
|
|
|
|
let pagedRowIds: [Int64] = PagedData.pagedRowIdsForRelatedRowIds(
|
|
|
|
|
db,
|
|
|
|
|
tableName: next.key,
|
|
|
|
|
pagedTableName: pagedTableName,
|
|
|
|
|
relatedRowIds: Array(next.value.map { $0.rowId }.asSet()),
|
|
|
|
|
joinToPagedType: joinToPagedType
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
result.append(contentsOf: pagedRowIds)
|
|
|
|
|
}
|
|
|
|
|
.asSet()
|
|
|
|
|
}()
|
|
|
|
|
|
|
|
|
|
guard !changesToQuery.isEmpty || !pagedRowIdsForRelatedChanges.isEmpty || !relatedDeletions.isEmpty else {
|
|
|
|
|
let associatedData: AssociatedDataInfo = getAssociatedDataInfo(db, updatedPageInfo)
|
|
|
|
|
return (updatedDataCache, updatedPageInfo, !deletionChanges.isEmpty, associatedData)
|
|
|
|
|
}
|
|
|
|
|
.asSet()
|
|
|
|
|
}()
|
|
|
|
|
|
|
|
|
|
guard !changesToQuery.isEmpty || !pagedRowIdsForRelatedChanges.isEmpty || !relatedDeletions.isEmpty else {
|
|
|
|
|
updateDataAndCallbackIfNeeded(updatedDataCache, updatedPageInfo, !deletionChanges.isEmpty)
|
|
|
|
|
return
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// Fetch the indexes of the rowIds so we can determine whether they should be added to the screen
|
|
|
|
|
let directRowIds: Set<Int64> = changesToQuery.map { $0.rowId }.asSet()
|
|
|
|
|
let pagedRowIdsForRelatedDeletions: Set<Int64> = relatedDeletions
|
|
|
|
|
.compactMap { $0.pagedRowIdsForRelatedDeletion }
|
|
|
|
|
.flatMap { $0 }
|
|
|
|
|
.asSet()
|
|
|
|
|
let itemIndexes: [PagedData.RowIndexInfo] = PagedData.indexes(
|
|
|
|
|
db,
|
|
|
|
|
rowIds: Array(directRowIds),
|
|
|
|
|
tableName: pagedTableName,
|
|
|
|
|
requiredJoinSQL: joinSQL,
|
|
|
|
|
orderSQL: orderSQL,
|
|
|
|
|
filterSQL: filterSQL
|
|
|
|
|
)
|
|
|
|
|
let relatedChangeIndexes: [PagedData.RowIndexInfo] = PagedData.indexes(
|
|
|
|
|
db,
|
|
|
|
|
rowIds: Array(pagedRowIdsForRelatedChanges),
|
|
|
|
|
tableName: pagedTableName,
|
|
|
|
|
requiredJoinSQL: joinSQL,
|
|
|
|
|
orderSQL: orderSQL,
|
|
|
|
|
filterSQL: filterSQL
|
|
|
|
|
)
|
|
|
|
|
let relatedDeletionIndexes: [PagedData.RowIndexInfo] = PagedData.indexes(
|
|
|
|
|
db,
|
|
|
|
|
rowIds: Array(pagedRowIdsForRelatedDeletions),
|
|
|
|
|
tableName: pagedTableName,
|
|
|
|
|
requiredJoinSQL: joinSQL,
|
|
|
|
|
orderSQL: orderSQL,
|
|
|
|
|
filterSQL: filterSQL
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
// Determine if the indexes for the row ids should be displayed on the screen and remove any
|
|
|
|
|
// which shouldn't - values less than 'currentCount' or if there is at least one value less than
|
|
|
|
|
// 'currentCount' and the indexes are sequential (ie. more than the current loaded content was
|
|
|
|
|
// added at once)
|
|
|
|
|
func determineValidChanges(for indexInfo: [PagedData.RowIndexInfo]) -> [Int64] {
|
|
|
|
|
let indexes: [Int64] = Array(indexInfo
|
|
|
|
|
.map { $0.rowIndex }
|
|
|
|
|
.sorted()
|
|
|
|
|
.asSet())
|
|
|
|
|
let indexesAreSequential: Bool = (indexes.map { $0 - 1 }.dropFirst() == indexes.dropLast())
|
|
|
|
|
let hasOneValidIndex: Bool = indexInfo.contains(where: { info -> Bool in
|
|
|
|
|
info.rowIndex >= updatedPageInfo.pageOffset && (
|
|
|
|
|
info.rowIndex < updatedPageInfo.currentCount || (
|
|
|
|
|
updatedPageInfo.currentCount < updatedPageInfo.pageSize &&
|
|
|
|
|
info.rowIndex <= (updatedPageInfo.pageOffset + updatedPageInfo.pageSize)
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
// Fetch the indexes of the rowIds so we can determine whether they should be added to the screen
|
|
|
|
|
let directRowIds: Set<Int64> = changesToQuery.map { $0.rowId }.asSet()
|
|
|
|
|
let pagedRowIdsForRelatedDeletions: Set<Int64> = relatedDeletions
|
|
|
|
|
.compactMap { $0.pagedRowIdsForRelatedDeletion }
|
|
|
|
|
.flatMap { $0 }
|
|
|
|
|
.asSet()
|
|
|
|
|
let itemIndexes: [PagedData.RowIndexInfo] = PagedData.indexes(
|
|
|
|
|
db,
|
|
|
|
|
rowIds: Array(directRowIds),
|
|
|
|
|
tableName: pagedTableName,
|
|
|
|
|
requiredJoinSQL: joinSQL,
|
|
|
|
|
orderSQL: orderSQL,
|
|
|
|
|
filterSQL: filterSQL
|
|
|
|
|
)
|
|
|
|
|
})
|
|
|
|
|
|
|
|
|
|
return (indexesAreSequential && hasOneValidIndex ?
|
|
|
|
|
indexInfo.map { $0.rowId } :
|
|
|
|
|
indexInfo
|
|
|
|
|
.filter { info -> Bool in
|
|
|
|
|
let relatedChangeIndexes: [PagedData.RowIndexInfo] = PagedData.indexes(
|
|
|
|
|
db,
|
|
|
|
|
rowIds: Array(pagedRowIdsForRelatedChanges),
|
|
|
|
|
tableName: pagedTableName,
|
|
|
|
|
requiredJoinSQL: joinSQL,
|
|
|
|
|
orderSQL: orderSQL,
|
|
|
|
|
filterSQL: filterSQL
|
|
|
|
|
)
|
|
|
|
|
let relatedDeletionIndexes: [PagedData.RowIndexInfo] = PagedData.indexes(
|
|
|
|
|
db,
|
|
|
|
|
rowIds: Array(pagedRowIdsForRelatedDeletions),
|
|
|
|
|
tableName: pagedTableName,
|
|
|
|
|
requiredJoinSQL: joinSQL,
|
|
|
|
|
orderSQL: orderSQL,
|
|
|
|
|
filterSQL: filterSQL
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
// Determine if the indexes for the row ids should be displayed on the screen and remove any
|
|
|
|
|
// which shouldn't - values less than 'currentCount' or if there is at least one value less than
|
|
|
|
|
// 'currentCount' and the indexes are sequential (ie. more than the current loaded content was
|
|
|
|
|
// added at once)
|
|
|
|
|
func determineValidChanges(for indexInfo: [PagedData.RowIndexInfo]) -> [Int64] {
|
|
|
|
|
let indexes: [Int64] = Array(indexInfo
|
|
|
|
|
.map { $0.rowIndex }
|
|
|
|
|
.sorted()
|
|
|
|
|
.asSet())
|
|
|
|
|
let indexesAreSequential: Bool = (indexes.map { $0 - 1 }.dropFirst() == indexes.dropLast())
|
|
|
|
|
let hasOneValidIndex: Bool = indexInfo.contains(where: { info -> Bool in
|
|
|
|
|
info.rowIndex >= updatedPageInfo.pageOffset && (
|
|
|
|
|
info.rowIndex < updatedPageInfo.currentCount || (
|
|
|
|
|
updatedPageInfo.currentCount < updatedPageInfo.pageSize &&
|
|
|
|
|
info.rowIndex <= (updatedPageInfo.pageOffset + updatedPageInfo.pageSize)
|
|
|
|
|
)
|
|
|
|
|
)
|
|
|
|
|
})
|
|
|
|
|
|
|
|
|
|
return (indexesAreSequential && hasOneValidIndex ?
|
|
|
|
|
indexInfo.map { $0.rowId } :
|
|
|
|
|
indexInfo
|
|
|
|
|
.filter { info -> Bool in
|
|
|
|
|
info.rowIndex >= updatedPageInfo.pageOffset && (
|
|
|
|
|
info.rowIndex < updatedPageInfo.currentCount || (
|
|
|
|
|
updatedPageInfo.currentCount < updatedPageInfo.pageSize &&
|
|
|
|
|
info.rowIndex <= (updatedPageInfo.pageOffset + updatedPageInfo.pageSize)
|
|
|
|
|
)
|
|
|
|
|
)
|
|
|
|
|
}
|
|
|
|
|
.map { info -> Int64 in info.rowId }
|
|
|
|
|
)
|
|
|
|
|
}
|
|
|
|
|
let validChangeRowIds: [Int64] = determineValidChanges(for: itemIndexes)
|
|
|
|
|
let validRelatedChangeRowIds: [Int64] = determineValidChanges(for: relatedChangeIndexes)
|
|
|
|
|
let validRelatedDeletionRowIds: [Int64] = determineValidChanges(for: relatedDeletionIndexes)
|
|
|
|
|
let countBefore: Int = itemIndexes.filter { $0.rowIndex < updatedPageInfo.pageOffset }.count
|
|
|
|
|
|
|
|
|
|
// If the number of indexes doesn't match the number of rowIds then it means something changed
|
|
|
|
|
// resulting in an item being filtered out
|
|
|
|
|
func performRemovalsIfNeeded(for rowIds: Set<Int64>, indexes: [PagedData.RowIndexInfo]) {
|
|
|
|
|
let uniqueIndexes: Set<Int64> = indexes.map { $0.rowId }.asSet()
|
|
|
|
|
|
|
|
|
|
// If they have the same count then nothin was filtered out so do nothing
|
|
|
|
|
guard rowIds.count != uniqueIndexes.count else { return }
|
|
|
|
|
|
|
|
|
|
// Otherwise something was probably removed so try to remove it from the cache
|
|
|
|
|
let rowIdsRemoved: Set<Int64> = rowIds.subtracting(uniqueIndexes)
|
|
|
|
|
let preDeletionCount: Int = updatedDataCache.count
|
|
|
|
|
updatedDataCache = updatedDataCache.deleting(rowIds: Array(rowIdsRemoved))
|
|
|
|
|
|
|
|
|
|
// Lastly make sure there were actually changes before updating the page info
|
|
|
|
|
guard updatedDataCache.count != preDeletionCount else { return }
|
|
|
|
|
|
|
|
|
|
let dataSizeDiff: Int = (updatedDataCache.count - preDeletionCount)
|
|
|
|
|
|
|
|
|
|
updatedPageInfo = PagedData.PageInfo(
|
|
|
|
|
pageSize: updatedPageInfo.pageSize,
|
|
|
|
|
pageOffset: updatedPageInfo.pageOffset,
|
|
|
|
|
currentCount: (updatedPageInfo.currentCount + dataSizeDiff),
|
|
|
|
|
totalCount: (updatedPageInfo.totalCount + dataSizeDiff)
|
|
|
|
|
)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// Actually perform any required removals
|
|
|
|
|
performRemovalsIfNeeded(for: directRowIds, indexes: itemIndexes)
|
|
|
|
|
performRemovalsIfNeeded(for: pagedRowIdsForRelatedChanges, indexes: relatedChangeIndexes)
|
|
|
|
|
performRemovalsIfNeeded(for: pagedRowIdsForRelatedDeletions, indexes: relatedDeletionIndexes)
|
|
|
|
|
|
|
|
|
|
// Update the offset and totalCount even if the rows are outside of the current page (need to
|
|
|
|
|
// in order to ensure the 'load more' sections are accurate)
|
|
|
|
|
updatedPageInfo = PagedData.PageInfo(
|
|
|
|
|
pageSize: updatedPageInfo.pageSize,
|
|
|
|
|
pageOffset: (updatedPageInfo.pageOffset + countBefore),
|
|
|
|
|
currentCount: updatedPageInfo.currentCount,
|
|
|
|
|
totalCount: (
|
|
|
|
|
updatedPageInfo.totalCount +
|
|
|
|
|
changesToQuery
|
|
|
|
|
.filter { $0.kind == .insert }
|
|
|
|
|
.filter { validChangeRowIds.contains($0.rowId) }
|
|
|
|
|
.count
|
|
|
|
|
)
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
// If there are no valid row ids then early-out (at this point the pageInfo would have changed
|
|
|
|
|
// so we want to flat 'hasChanges' as true)
|
|
|
|
|
guard !validChangeRowIds.isEmpty || !validRelatedChangeRowIds.isEmpty || !validRelatedDeletionRowIds.isEmpty else {
|
|
|
|
|
let associatedData: AssociatedDataInfo = getAssociatedDataInfo(db, updatedPageInfo)
|
|
|
|
|
return (updatedDataCache, updatedPageInfo, true, associatedData)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// Fetch the inserted/updated rows
|
|
|
|
|
let targetRowIds: [Int64] = Array((validChangeRowIds + validRelatedChangeRowIds + validRelatedDeletionRowIds).asSet())
|
|
|
|
|
let updatedItems: [T] = {
|
|
|
|
|
do { return try dataQuery(targetRowIds).fetchAll(db) }
|
|
|
|
|
catch {
|
|
|
|
|
SNLog("[PagedDatabaseObserver] Error fetching data during change: \(error)")
|
|
|
|
|
return []
|
|
|
|
|
}
|
|
|
|
|
.map { info -> Int64 in info.rowId }
|
|
|
|
|
)
|
|
|
|
|
}()
|
|
|
|
|
|
|
|
|
|
updatedDataCache = updatedDataCache.upserting(items: updatedItems)
|
|
|
|
|
|
|
|
|
|
// Update the currentCount for the upserted data
|
|
|
|
|
let dataSizeDiff: Int = (updatedDataCache.count - oldDataCount)
|
|
|
|
|
updatedPageInfo = PagedData.PageInfo(
|
|
|
|
|
pageSize: updatedPageInfo.pageSize,
|
|
|
|
|
pageOffset: updatedPageInfo.pageOffset,
|
|
|
|
|
currentCount: (updatedPageInfo.currentCount + dataSizeDiff),
|
|
|
|
|
totalCount: updatedPageInfo.totalCount
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
// Return the final updated data
|
|
|
|
|
let associatedData: AssociatedDataInfo = getAssociatedDataInfo(db, updatedPageInfo)
|
|
|
|
|
return (updatedDataCache, updatedPageInfo, true, associatedData)
|
|
|
|
|
}
|
|
|
|
|
.defaulting(to: (cache: dataCache, pageInfo: pageInfo, hasChanges: false, associatedData: []))
|
|
|
|
|
|
|
|
|
|
// Now that we have all of the changes, check if there were actually any changes
|
|
|
|
|
guard updatedData.hasChanges || updatedData.associatedData.contains(where: { hasChanges, _ in hasChanges }) else {
|
|
|
|
|
return
|
|
|
|
|
}
|
|
|
|
|
let validChangeRowIds: [Int64] = determineValidChanges(for: itemIndexes)
|
|
|
|
|
let validRelatedChangeRowIds: [Int64] = determineValidChanges(for: relatedChangeIndexes)
|
|
|
|
|
let validRelatedDeletionRowIds: [Int64] = determineValidChanges(for: relatedDeletionIndexes)
|
|
|
|
|
let countBefore: Int = itemIndexes.filter { $0.rowIndex < updatedPageInfo.pageOffset }.count
|
|
|
|
|
|
|
|
|
|
// If the number of indexes doesn't match the number of rowIds then it means something changed
|
|
|
|
|
// resulting in an item being filtered out
|
|
|
|
|
func performRemovalsIfNeeded(for rowIds: Set<Int64>, indexes: [PagedData.RowIndexInfo]) {
|
|
|
|
|
let uniqueIndexes: Set<Int64> = indexes.map { $0.rowId }.asSet()
|
|
|
|
|
|
|
|
|
|
// If they have the same count then nothin was filtered out so do nothing
|
|
|
|
|
guard rowIds.count != uniqueIndexes.count else { return }
|
|
|
|
|
|
|
|
|
|
// Otherwise something was probably removed so try to remove it from the cache
|
|
|
|
|
let rowIdsRemoved: Set<Int64> = rowIds.subtracting(uniqueIndexes)
|
|
|
|
|
let preDeletionCount: Int = updatedDataCache.count
|
|
|
|
|
updatedDataCache = updatedDataCache.deleting(rowIds: Array(rowIdsRemoved))
|
|
|
|
|
// If the associated data changed then update the updatedCachedData with the updated associated data
|
|
|
|
|
var finalUpdatedDataCache: DataCache<T> = updatedData.cache
|
|
|
|
|
|
|
|
|
|
// Lastly make sure there were actually changes before updating the page info
|
|
|
|
|
guard updatedDataCache.count != preDeletionCount else { return }
|
|
|
|
|
|
|
|
|
|
let dataSizeDiff: Int = (updatedDataCache.count - preDeletionCount)
|
|
|
|
|
updatedData.associatedData.forEach { hasChanges, associatedData in
|
|
|
|
|
guard updatedData.hasChanges || hasChanges else { return }
|
|
|
|
|
|
|
|
|
|
updatedPageInfo = PagedData.PageInfo(
|
|
|
|
|
pageSize: updatedPageInfo.pageSize,
|
|
|
|
|
pageOffset: updatedPageInfo.pageOffset,
|
|
|
|
|
currentCount: (updatedPageInfo.currentCount + dataSizeDiff),
|
|
|
|
|
totalCount: (updatedPageInfo.totalCount + dataSizeDiff)
|
|
|
|
|
)
|
|
|
|
|
finalUpdatedDataCache = associatedData.updateAssociatedData(to: finalUpdatedDataCache)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// Actually perform any required removals
|
|
|
|
|
performRemovalsIfNeeded(for: directRowIds, indexes: itemIndexes)
|
|
|
|
|
performRemovalsIfNeeded(for: pagedRowIdsForRelatedChanges, indexes: relatedChangeIndexes)
|
|
|
|
|
performRemovalsIfNeeded(for: pagedRowIdsForRelatedDeletions, indexes: relatedDeletionIndexes)
|
|
|
|
|
|
|
|
|
|
// Update the offset and totalCount even if the rows are outside of the current page (need to
|
|
|
|
|
// in order to ensure the 'load more' sections are accurate)
|
|
|
|
|
updatedPageInfo = PagedData.PageInfo(
|
|
|
|
|
pageSize: updatedPageInfo.pageSize,
|
|
|
|
|
pageOffset: (updatedPageInfo.pageOffset + countBefore),
|
|
|
|
|
currentCount: updatedPageInfo.currentCount,
|
|
|
|
|
totalCount: (
|
|
|
|
|
updatedPageInfo.totalCount +
|
|
|
|
|
changesToQuery
|
|
|
|
|
.filter { $0.kind == .insert }
|
|
|
|
|
.filter { validChangeRowIds.contains($0.rowId) }
|
|
|
|
|
.count
|
|
|
|
|
)
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
// If there are no valid row ids then stop here (trigger updates though since the page info
|
|
|
|
|
// has changes)
|
|
|
|
|
guard !validChangeRowIds.isEmpty || !validRelatedChangeRowIds.isEmpty || !validRelatedDeletionRowIds.isEmpty else {
|
|
|
|
|
updateDataAndCallbackIfNeeded(updatedDataCache, updatedPageInfo, true)
|
|
|
|
|
return
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// Fetch the inserted/updated rows
|
|
|
|
|
let targetRowIds: [Int64] = Array((validChangeRowIds + validRelatedChangeRowIds + validRelatedDeletionRowIds).asSet())
|
|
|
|
|
let updatedItems: [T] = (try? dataQuery(targetRowIds)
|
|
|
|
|
.fetchAll(db))
|
|
|
|
|
.defaulting(to: [])
|
|
|
|
|
// Update the cache, pageInfo and the change callback
|
|
|
|
|
self.dataCache.mutate { $0 = finalUpdatedDataCache }
|
|
|
|
|
self.pageInfo.mutate { $0 = updatedData.pageInfo }
|
|
|
|
|
|
|
|
|
|
// Process the upserted data
|
|
|
|
|
updatedDataCache = updatedDataCache.upserting(items: updatedItems)
|
|
|
|
|
|
|
|
|
|
// Update the currentCount for the upserted data
|
|
|
|
|
let dataSizeDiff: Int = (updatedDataCache.count - oldDataCount)
|
|
|
|
|
|
|
|
|
|
updatedPageInfo = PagedData.PageInfo(
|
|
|
|
|
pageSize: updatedPageInfo.pageSize,
|
|
|
|
|
pageOffset: updatedPageInfo.pageOffset,
|
|
|
|
|
currentCount: (updatedPageInfo.currentCount + dataSizeDiff),
|
|
|
|
|
totalCount: updatedPageInfo.totalCount
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
updateDataAndCallbackIfNeeded(updatedDataCache, updatedPageInfo, true)
|
|
|
|
|
// Trigger the unsorted change callback (the actual UI update triggering should eventually be run on
|
|
|
|
|
// the main thread via the `PagedData.processAndTriggerUpdates` function)
|
|
|
|
|
self.onChangeUnsorted(finalUpdatedDataCache.values, updatedData.pageInfo)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
public func databaseDidRollback(_ db: Database) {}
|
|
|
|
|
|
Ryan Zhao
11 months ago