rxdb-db-collection fixes (#667)

* Refactor types and cleanup in rxdb.ts

Replaces generic 'any' types with more specific types such as 'Subscription' and 'Record<string, unknown>' for improved type safety. Removes unnecessary type assertions and updates function signatures and variable declarations accordingly.

* Filter out deleted documents in sync queries

Added '_deleted': false to query selectors to ensure only non-deleted documents are included during synchronization. This improves data consistency by excluding soft-deleted records from sync operations.

* Refactor stripRxdbFields function signature

Updated the stripRxdbFields function to use a stricter type for the input parameter and removed unnecessary handling for arrays and nullish values.

* Add rxjs as peer dependency

Added 'rxjs' with version >=7.8.2 to peerDependencies in package.json to ensure compatibility and proper dependency management.

* changeset

---------

Co-authored-by: Sam Willis <sam.willis@gmail.com>

Author: daveycodez

.changeset/smooth-candies-visit.md

@@ -0,0 +1,5 @@
+---
+"@tanstack/rxdb-db-collection": patch
+---
+
+Addressed the majority of "any" type usage in the rxdb-db-collection adapter as well as fixed querying for documents where \_deleted is false, which fixes issues when using localStorage or other persistent storage mechanisms.

packages/rxdb-db-collection/package.json

@@ -35,6 +35,7 @@
   "packageManager": "pnpm@10.18.2",
   "peerDependencies": {
     "rxdb": ">=16.17.2",
+    "rxjs": ">=7.8.2",
     "typescript": ">=4.7"
   },
   "author": "Kyle Mathews",

packages/rxdb-db-collection/src/helper.ts

@@ -5,11 +5,8 @@ const RESERVED_RXDB_FIELDS = new Set([
   `_meta`,
 ])
 
-export function stripRxdbFields<T extends Record<string, any>>(
-  obj: T | any
-): T {
-  if (!obj) return obj
-  const out: any = Array.isArray(obj) ? [] : {}
+export function stripRxdbFields<T extends Record<string, unknown>>(obj: T): T {
+  const out: Record<string, unknown> = {}
   for (const k of Object.keys(obj)) {
     if (RESERVED_RXDB_FIELDS.has(k)) continue
     out[k] = obj[k]

packages/rxdb-db-collection/src/rxdb.ts

@@ -13,6 +13,7 @@ import type {
   RxCollection,
   RxDocumentData,
 } from "rxdb/plugins/core"
+import type { Subscription } from "rxjs"
 
 import type {
   BaseCollectionConfig,
@@ -27,7 +28,10 @@ const debug = DebugModule.debug(`ts/db:rxdb`)
 /**
  * Used in tests to ensure proper cleanup
  */
-export const OPEN_RXDB_SUBSCRIPTIONS = new WeakMap<RxCollection, Set<any>>()
+export const OPEN_RXDB_SUBSCRIPTIONS = new WeakMap<
+  RxCollection,
+  Set<Subscription>
+>()
 
 /**
  * Configuration interface for RxDB collection options
@@ -106,7 +110,7 @@ export function rxdbCollectionOptions(config: RxDBCollectionConfig<any, any>) {
 
   // "getKey"
   const primaryPath = rxCollection.schema.primaryPath
-  function getKey(item: any): string {
+  function getKey(item: Record<string, unknown>): string {
     const key: string = item[primaryPath] as string
     return key
   }
@@ -147,15 +151,16 @@ export function rxdbCollectionOptions(config: RxDBCollectionConfig<any, any>) {
                     },
                   },
                 ],
-              } as any,
-              sort: [{ "_meta.lwt": `asc` }, { [primaryPath]: `asc` } as any],
+                _deleted: false,
+              },
+              sort: [{ "_meta.lwt": `asc` }, { [primaryPath]: `asc` }],
               limit: syncBatchSize,
               skip: 0,
             }
           } else {
             query = {
-              selector: {},
-              sort: [{ "_meta.lwt": `asc` }, { [primaryPath]: `asc` } as any],
+              selector: { _deleted: false },
+              sort: [{ "_meta.lwt": `asc` }, { [primaryPath]: `asc` }],
               limit: syncBatchSize,
               skip: 0,
             }
@@ -198,11 +203,11 @@ export function rxdbCollectionOptions(config: RxDBCollectionConfig<any, any>) {
           return
         }
         begin()
-        write(msg as any)
+        write(msg)
         commit()
       }
 
-      let sub: any
+      let sub: Subscription
       function startOngoingFetch() {
         // Subscribe early and buffer live changes during initial load and ongoing
         sub = rxCollection.$.subscribe((ev) => {
@@ -234,7 +239,7 @@ export function rxdbCollectionOptions(config: RxDBCollectionConfig<any, any>) {
 
         if (buffer.length) {
           begin()
-          for (const msg of buffer) write(msg as any)
+          for (const msg of buffer) write(msg)
           commit()
           buffer.length = 0
         }
@@ -258,14 +263,14 @@ export function rxdbCollectionOptions(config: RxDBCollectionConfig<any, any>) {
     getSyncMetadata: undefined,
   }
 
-  const collectionConfig: CollectionConfig<Row, string, any> = {
+  const collectionConfig: CollectionConfig<Row, string> = {
     ...restConfig,
-    getKey: getKey as any,
+    getKey,
     sync,
     onInsert: async (params) => {
       debug(`insert`, params)
       const newItems = params.transaction.mutations.map((m) => m.modified)
-      return rxCollection.bulkUpsert(newItems as Array<any>).then((result) => {
+      return rxCollection.bulkUpsert(newItems).then((result) => {
         if (result.error.length > 0) {
           throw rxStorageWriteErrorToRxError(ensureNotFalsy(result.error[0]))
         }
@@ -285,15 +290,15 @@ export function rxdbCollectionOptions(config: RxDBCollectionConfig<any, any>) {
         if (!doc) {
           continue
         }
-        await doc.incrementalPatch(newValue as any)
+        await doc.incrementalPatch(newValue)
       }
     },
     onDelete: async (params) => {
       debug(`delete`, params)
       const mutations = params.transaction.mutations.filter(
         (m) => m.type === `delete`
       )
-      const ids = mutations.map((mutation: any) => getKey(mutation.original))
+      const ids = mutations.map((mutation) => getKey(mutation.original))
       return rxCollection.bulkRemove(ids).then((result) => {
         if (result.error.length > 0) {
           throw result.error