add pagination

src/collection-replication-state.ts

@@ -58,6 +58,12 @@ export class CollectionReplicationState<T extends RxCollection> {
 	readonly lastModified$: Observable<string> = this.subjects.lastModified.asObservable();
 	readonly paused$: Observable<boolean> = this.subjects.paused.asObservable();
 
+	/**
+	 *
+	 */
+	private firstSyncResolver: (() => void) | null = null;
+	public readonly firstSync: Promise<void>;
+
 	/**
 	 *
 	 */
@@ -67,6 +73,11 @@ export class CollectionReplicationState<T extends RxCollection> {
 		this.hooks = hooks || {};
 		this.endpoint = endpoint;
 
+		// Initialize the firstSync promise
+		this.firstSync = new Promise<void>((resolve) => {
+			this.firstSyncResolver = resolve;
+		});
+
 		/**
 		 * Subscribe to the remoteIDs local document
 		 */
@@ -126,8 +137,11 @@ export class CollectionReplicationState<T extends RxCollection> {
 	 */
 	async run() {
 		await this.auditIDs();
-		// await this.syncLastModified();
-		// await this.syncRemoteIDs();
+
+		if (this.firstSyncResolver) {
+			this.firstSyncResolver();
+			this.firstSyncResolver = null; // Clear the resolver to prevent future calls
+		}
 	}
 
 	/**

src/manager.ts

@@ -127,20 +127,38 @@ export class Manager<TDatabase extends RxDatabase> {
 				collectionReplication.start();
 
 				/**
-				 * Subscribe to query params and register a new replication state for the query
-				 * - also cancel the previous query replication
+				 * Add some subscriptions for the Query Class
+				 * - these will be completed when the query instance is cancelled
 				 */
-				this.subs.push(
+				query.subs.push(
+					/**
+					 * Subscribe to query params and register a new replication state for the query
+					 * - also cancel the previous query replication
+					 */
 					query.params$.subscribe((params) => {
 						const apiQueryParams = this.getApiQueryParams(params);
 						const queryEndpoint = buildUrlWithParams(endpoint, apiQueryParams);
-						const queryReplication = this.registerQueryReplication({
-							collectionReplication,
-							collection,
-							endpoint: queryEndpoint,
-						});
-						this.addQueryKeyToReplicationsMap(key, queryEndpoint);
-						queryReplication.start();
+
+						if (!this.replicationStates.has(queryEndpoint)) {
+							const queryReplication = this.registerQueryReplication({
+								collectionReplication,
+								collection,
+								endpoint: queryEndpoint,
+							});
+
+							this.addQueryKeyToReplicationsMap(key, queryEndpoint);
+
+							/**
+							 * Subscribe to the query trigger and trigger the query replication
+							 */
+							query.subs.push(
+								query.triggerServerQuery$.subscribe((page) => {
+									queryReplication.nextPage();
+								})
+							);
+
+							queryReplication.start();
+						}
 					})
 				);
 
@@ -238,27 +256,24 @@ export class Manager<TDatabase extends RxDatabase> {
 	 * There is one replication state per unique query
 	 */
 	registerQueryReplication({ endpoint, collectionReplication, collection }) {
-		if (!this.replicationStates.has(endpoint)) {
-			const queryReplication = new QueryReplicationState({
-				httpClient: this.httpClient,
-				collectionReplication,
-				collection,
-				endpoint,
-			});
-
-			/**
-			 * Subscribe to query errors and pipe them to the error subject
-			 */
-			this.subs.push(
-				queryReplication.error$.subscribe((error) => {
-					this.subjects.error.next(error);
-				})
-			);
+		const queryReplication = new QueryReplicationState({
+			httpClient: this.httpClient,
+			collectionReplication,
+			collection,
+			endpoint,
+		});
 
-			this.replicationStates.set(endpoint, queryReplication);
-		}
+		/**
+		 * Subscribe to query errors and pipe them to the error subject
+		 */
+		this.subs.push(
+			queryReplication.error$.subscribe((error) => {
+				this.subjects.error.next(error);
+			})
+		);
 
-		return this.replicationStates.get(endpoint);
+		this.replicationStates.set(endpoint, collectionReplication);
+		return queryReplication;
 	}
 
 	/**

src/query-replication-state.ts

@@ -36,15 +36,19 @@ export class QueryReplicationState<T extends RxCollection> {
 	public readonly subjects = {
 		error: new Subject<Error>(),
 		paused: new BehaviorSubject<boolean>(true), // true when the replication is paused, start true
+		active: new BehaviorSubject<boolean>(false), // true when something is running, false when not
 	};
 
-	readonly paused$: Observable<boolean> = this.subjects.paused.asObservable();
-
 	/**
 	 *
 	 */
 	readonly error$: Observable<Error> = this.subjects.error.asObservable();
+	readonly paused$: Observable<boolean> = this.subjects.paused.asObservable();
+	readonly active$: Observable<boolean> = this.subjects.active.asObservable();
 
+	/**
+	 *
+	 */
 	constructor({
 		collection,
 		httpClient,
@@ -60,6 +64,9 @@ export class QueryReplicationState<T extends RxCollection> {
 		 *
 		 */
 		this.subs.push(
+			/**
+			 * Pause/Start the replication
+			 */
 			this.paused$
 				.pipe(
 					switchMap((isPaused) => (isPaused ? [] : interval(this.pollingTime).pipe(startWith(0)))),
@@ -68,36 +75,47 @@ export class QueryReplicationState<T extends RxCollection> {
 				.subscribe(async () => {
 					this.run();
 				})
+
+			/**
+			 *
+			 */
 		);
 	}
 
+	/**
+	 *
+	 */
 	async run() {
+		await this.collectionReplication.firstSync;
 		await this.runPull();
 	}
 
+	/**
+	 *
+	 */
 	async runPull() {
-		if (this.isStopped()) {
+		if (this.isStopped() || this.subjects.active.getValue()) {
+			return;
+		}
+
+		this.subjects.active.next(true);
+		const include = this.collectionReplication.getUnsyncedRemoteIDs();
+		const lastModified = this.collectionReplication.subjects.lastModified.getValue();
+
+		/**
+		 * If there is nothing to fetch, then return
+		 */
+		if (isEmpty(include) && !lastModified) {
 			return;
 		}
 
 		try {
 			let response;
-			const include = this.collectionReplication.getUnsyncedRemoteIDs();
 
 			if (isEmpty(include)) {
-				response = await this.httpClient.get(this.endpoint);
+				response = await this.fetchLastModified({ lastModified });
 			} else {
-				response = await this.httpClient.post(
-					this.endpoint,
-					{
-						include,
-					},
-					{
-						headers: {
-							'X-HTTP-Method-Override': 'GET',
-						},
-					}
-				);
+				response = await this.fetchRemoteIDs({ include });
 			}
 
 			if (!response.data) {
@@ -115,9 +133,50 @@ export class QueryReplicationState<T extends RxCollection> {
 			await this.collection.bulkUpsert(documents);
 		} catch (error) {
 			this.subjects.error.next(error);
+		} finally {
+			this.subjects.active.next(false);
 		}
 	}
 
+	/**
+	 *
+	 */
+	async fetchRemoteIDs({ include }) {
+		const response = await this.httpClient.post(
+			this.endpoint,
+			{
+				include,
+			},
+			{
+				headers: {
+					'X-HTTP-Method-Override': 'GET',
+				},
+			}
+		);
+
+		return response;
+	}
+
+	/**
+	 *
+	 */
+	async fetchLastModified({ lastModified }) {
+		const response = await this.httpClient.get(this.endpoint, {
+			params: {
+				modified_after: lastModified,
+			},
+		});
+
+		return response;
+	}
+
+	/**
+	 *
+	 */
+	nextPage() {
+		this.run();
+	}
+
 	/**
 	 * We need to a way to pause and start the replication, eg: when the user is offline
 	 */

src/query-state.ts

@@ -43,6 +43,8 @@ export class Query<T extends RxCollection> {
 	private isCanceled = false;
 	private whereClauses: WhereClause[] = [];
 	private hooks: QueryConfig<T>['hooks'];
+	private paginationEndReached = false;
+	private pageSize: number;
 
 	/**
 	 *
@@ -53,6 +55,9 @@ export class Query<T extends RxCollection> {
 		result: new BehaviorSubject<DocumentType<T>[]>([]),
 		error: new Subject<Error>(),
 		cancel: new Subject<void>(),
+		currentPage: new BehaviorSubject<number>(1),
+		paginatedResult: new BehaviorSubject<DocumentType<T>[]>([]),
+		triggerServerQuery: new Subject<void>(),
 	};
 
 	/**
@@ -62,9 +67,13 @@ export class Query<T extends RxCollection> {
 	readonly result$: Observable<DocumentType<T>[]> = this.subjects.result.asObservable();
 	readonly error$: Observable<Error> = this.subjects.error.asObservable();
 	readonly cancel$: Observable<void> = this.subjects.cancel.asObservable();
+	readonly currentPage$: Observable<number> = this.subjects.currentPage.asObservable();
+	readonly paginatedResult$: Observable<DocumentType<T>[]> =
+		this.subjects.paginatedResult.asObservable();
+	readonly triggerServerQuery$: Observable<void> = this.subjects.triggerServerQuery.asObservable();
 
 	readonly resource = new ObservableResource(this.result$);
-	readonly paginatedResource = new ObservableResource(this.result$);
+	readonly paginatedResource = new ObservableResource(this.paginatedResult$);
 
 	/**
 	 *
@@ -74,6 +83,7 @@ export class Query<T extends RxCollection> {
 		this.collection = collection;
 		this.subjects.params.next(initialParams);
 		this.hooks = hooks || {};
+		this.pageSize = 10;
 
 		/**
 		 * Keep track of what we are subscribed to
@@ -84,7 +94,25 @@ export class Query<T extends RxCollection> {
 			 */
 			this.find$.subscribe((result) => {
 				this.subjects.result.next(result);
-			})
+			}),
+
+			/**
+			 * Subscribe to result$ and emit paginated results
+			 */
+			this.result$
+				.pipe(
+					switchMap((items) =>
+						this.currentPage$.pipe(
+							map((currentPage) => {
+								const end = currentPage * this.pageSize;
+								const pageItems = items.slice(0, end);
+								this.paginationEndReached = pageItems.length < end;
+								return pageItems;
+							})
+						)
+					)
+				)
+				.subscribe(this.subjects.paginatedResult)
 		);
 	}
 
@@ -211,7 +239,20 @@ export class Query<T extends RxCollection> {
 	/**
 	 * Pagination
 	 */
-	nextPage() {}
+	nextPage() {
+		if (!this.paginationEndReached) {
+			const page = this.subjects.currentPage.value + 1;
+			this.subjects.currentPage.next(page);
+		} else {
+			// the Query Replication will listen for this event and trigger a server query
+			this.subjects.triggerServerQuery.next();
+		}
+	}
+
+	paginationReset() {
+		this.paginationEndReached = false;
+		this.subjects.currentPage.next(1);
+	}
 
 	/**
 	 * Cancel

src/use-query.ts

@@ -21,8 +21,8 @@ export const useQuery = (queryOptions: QueryOptions) => {
 
 	/**
 	 * This is a hack for when when collection is reset:
-	 * - re-register query
 	 * - re-render components that use this query
+	 * - on re-render the query is recreated
 	 */
 	const trigger = useObservableState(query.cancel$.pipe(map(() => trigger + 1)), 0);