forked from Nozbe/WatermelonDB
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[sync] Split sync/impl into smaller chunks
- Loading branch information
Showing
7 changed files
with
176 additions
and
142 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,68 @@ | ||
| // @flow | ||
|
|
||
| import { | ||
| // $FlowFixMe | ||
| promiseAllObject, | ||
| map, | ||
| reduce, | ||
| values, | ||
| pipe, | ||
| } from 'rambdax' | ||
| import { unnest } from '../../utils/fp' | ||
| import type { Database, Collection, Model } from '../..' | ||
| import * as Q from '../../QueryDescription' | ||
| import { columnName } from '../../Schema' | ||
|
|
||
| import type { SyncTableChangeSet, SyncDatabaseChangeSet } from '../index' | ||
| import { ensureActionsEnabled } from './helpers' | ||
|
|
||
| export type SyncLocalChanges = $Exact<{ changes: SyncDatabaseChangeSet, affectedRecords: Model[] }> | ||
|
|
||
| const notSyncedQuery = Q.where(columnName('_status'), Q.notEq('synced')) | ||
| // TODO: It would be best to omit _status, _changed fields, since they're not necessary for the server | ||
| // but this complicates markLocalChangesAsDone, since we don't have the exact copy to compare if record changed | ||
| // TODO: It would probably also be good to only send to server locally changed fields, not full records | ||
| const rawsForStatus = (status, records) => | ||
| reduce( | ||
| (raws, record) => (record._raw._status === status ? raws.concat({ ...record._raw }) : raws), | ||
| [], | ||
| records, | ||
| ) | ||
|
|
||
| async function fetchLocalChangesForCollection<T: Model>( | ||
| collection: Collection<T>, | ||
| ): Promise<[SyncTableChangeSet, T[]]> { | ||
| const changedRecords = await collection.query(notSyncedQuery).fetch() | ||
| const changeSet = { | ||
| created: rawsForStatus('created', changedRecords), | ||
| updated: rawsForStatus('updated', changedRecords), | ||
| deleted: await collection.database.adapter.getDeletedRecords(collection.table), | ||
| } | ||
| return [changeSet, changedRecords] | ||
| } | ||
|
|
||
| const extractChanges = map(([changeSet]) => changeSet) | ||
| const extractAllAffectedRecords = pipe( | ||
| values, | ||
| map(([, records]) => records), | ||
| unnest, | ||
| ) | ||
|
|
||
| export default function fetchLocalChanges(db: Database): Promise<SyncLocalChanges> { | ||
| ensureActionsEnabled(db) | ||
| return db.action(async () => { | ||
| const changes = await promiseAllObject( | ||
| map( | ||
| fetchLocalChangesForCollection, | ||
| // $FlowFixMe | ||
| db.collections.map, | ||
| ), | ||
| ) | ||
| // TODO: deep-freeze changes object (in dev mode only) to detect mutations (user bug) | ||
| return { | ||
| // $FlowFixMe | ||
| changes: extractChanges(changes), | ||
| affectedRecords: extractAllAffectedRecords(changes), | ||
| } | ||
| }, 'sync-fetchLocalChanges') | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,18 @@ | ||
| // @flow | ||
|
|
||
| import type { Database } from '../..' | ||
| import type { Timestamp } from '../index' | ||
|
|
||
| export { default as applyRemoteChanges } from './applyRemote' | ||
| export { default as fetchLocalChanges } from './fetchLocal' | ||
| export { default as markLocalChangesAsSynced } from './markAsSynced' | ||
|
|
||
| const lastSyncedAtKey = '__watermelon_last_pulled_at' | ||
|
|
||
| export async function getLastPulledAt(database: Database): Promise<?Timestamp> { | ||
| return parseInt(await database.adapter.getLocal(lastSyncedAtKey), 10) || null | ||
| } | ||
|
|
||
| export async function setLastPulledAt(database: Database, timestamp: Timestamp): Promise<void> { | ||
| await database.adapter.setLocal(lastSyncedAtKey, `${timestamp}`) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,70 @@ | ||
| // @flow | ||
|
|
||
| import { | ||
| // $FlowFixMe | ||
| promiseAllObject, | ||
| map, | ||
| reduce, | ||
| values, | ||
| pipe, | ||
| equals, | ||
| } from 'rambdax' | ||
| import { unnest } from '../../utils/fp' | ||
| import { logError } from '../../utils/common' | ||
| import type { Database, Model } from '../..' | ||
|
|
||
| import { prepareMarkAsSynced, ensureActionsEnabled } from './helpers' | ||
| import type { SyncLocalChanges } from './fetchLocal' | ||
|
|
||
| const unchangedRecordsForRaws = (raws, recordCache) => | ||
| reduce( | ||
| (records, raw) => { | ||
| const record = recordCache.find(model => model.id === raw.id) | ||
| if (!record) { | ||
| logError( | ||
| `[Sync] Looking for record ${ | ||
| raw.id | ||
| } to mark it as synced, but I can't find it. Will ignore it (it should get synced next time). This is probably a Watermelon bug — please file an issue!`, | ||
| ) | ||
| return records | ||
| } | ||
|
|
||
| // only include if it didn't change since fetch | ||
| // TODO: get rid of `equals` | ||
| return equals(record._raw, raw) ? records.concat(record) : records | ||
| }, | ||
| [], | ||
| raws, | ||
| ) | ||
|
|
||
| const recordsToMarkAsSynced = ({ changes, affectedRecords }: SyncLocalChanges): Model[] => | ||
| pipe( | ||
| values, | ||
| map(({ created, updated }) => | ||
| unchangedRecordsForRaws([...created, ...updated], affectedRecords), | ||
| ), | ||
| unnest, | ||
| )(changes) | ||
|
|
||
| const destroyDeletedRecords = (db: Database, { changes }: SyncLocalChanges): Promise<*> => | ||
| promiseAllObject( | ||
| map( | ||
| ({ deleted }, tableName) => db.adapter.destroyDeletedRecords(tableName, deleted), | ||
| // $FlowFixMe | ||
| changes, | ||
| ), | ||
| ) | ||
|
|
||
| export default function markLocalChangesAsSynced( | ||
| db: Database, | ||
| syncedLocalChanges: SyncLocalChanges, | ||
| ): Promise<void> { | ||
| ensureActionsEnabled(db) | ||
| return db.action(async () => { | ||
| // update and destroy records concurrently | ||
| await Promise.all([ | ||
| db.batch(...map(prepareMarkAsSynced, recordsToMarkAsSynced(syncedLocalChanges))), | ||
| destroyDeletedRecords(db, syncedLocalChanges), | ||
| ]) | ||
| }, 'sync-markLocalChangesAsSynced') | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters