Deleting RxDB rows does not persist to localStorage

[daveycodez]

I just made a very simple example where I am using only RxDB with TanStackDB, I'm able to add rows just fine. The only issue is when you delete a row, it deletes properly, but if you refresh the page, the row comes back, and it is impossible to delete it. This only occurs with localStorage, not memory adapter.

This must be an issue with creating an rxdbCollection, since it breaks as soon as you refresh the page and it brings back old deleted rows from RxDB. Maybe related to not correctly filtering out _deleted: true rows?

[daveycodez]

@kevin-dp

initialFetch() the query selector needs to have _deleted: false

let query: FilledMangoQuery<Row>
if (cursor) {
    query = {
        selector: {
            $or: [
                { "_meta.lwt": { $gt: cursor._meta.lwt } },
                {
                    "_meta.lwt": cursor._meta.lwt,
                    [primaryPath]: {
                        $gt: getKey(cursor),
                    },
                },
            ],
            _deleted: false,
        } as any,
        sort: [
            { "_meta.lwt": `asc` },
            { [primaryPath]: `asc` } as any,
        ],
        limit: syncBatchSize,
        skip: 0,
    }
} else {
    query = {
        selector: {
            _deleted: false,
        },
        sort: [
            { "_meta.lwt": `asc` },
            { [primaryPath]: `asc` } as any,
        ],
        limit: syncBatchSize,
        skip: 0,
    }
}

[daveycodez]

The other option is to address this ->

/**
 * Instead of doing a RxCollection.query(),
 * we directly query the storage engine of the RxCollection so we do not use the
 * RxCollection document cache because it likely wont be used anyway
 * since most queries will run directly on the tanstack-db side.
 */

If you skip RxCollection.find().exec() and use storageInstance directly, it will skip all the checks for deleted documents.

So another fix would be instead of checking for _deleted in the selector to do this:

const docs = await rxCollection
  .find(preparedQuery.query)
  .exec()

cursor = lastOfArray(docs)
if (docs.length === 0) {
    ready = true
    break
}

docs.forEach((d) => {
    write({
        type: `insert`,
        value: stripRxdbFields(clone(d.toJSON())),
    })
})

I probably prefer this method @KyleAMathews

[daveycodez]

I also cleaned up 99% of the "any" types.

rxdb.ts

import type { StandardSchemaV1 } from "@standard-schema/spec"
import type {
    BaseCollectionConfig,
    CollectionConfig,
    InferSchemaOutput,
    SyncConfig
} from "@tanstack/db"
import DebugModule from "debug"
import type {
    FilledMangoQuery,
    RxCollection,
    RxDocumentData
} from "rxdb/plugins/core"
import {
    clone,
    ensureNotFalsy,
    getFromMapOrCreate,
    lastOfArray,
    prepareQuery,
    rxStorageWriteErrorToRxError
} from "rxdb/plugins/core"
import type { Subscription } from "rxjs"
import { stripRxdbFields } from "./helper"

const debug = DebugModule.debug(`ts/db:rxdb`)

/**
 * Used in tests to ensure proper cleanup
 */
export const OPEN_RXDB_SUBSCRIPTIONS = new WeakMap<
    RxCollection,
    Set<Subscription>
>()

/**
 * Configuration interface for RxDB collection options
 * @template T - The explicit type of items in the collection (highest priority). Use the document type of your RxCollection here.
 * @template TSchema - The schema type for validation and type inference (second priority)
 *
 * @remarks
 * Type resolution follows a priority order:
 * 1. If you provide an explicit type via generic parameter, it will be used
 * 2. If no explicit type is provided but a schema is, the schema's output type will be inferred
 *
 * You should provide EITHER an explicit type OR a schema, but not both, as they would conflict.
 * Notice that primary keys in RxDB must always be a string.
 */
export type RxDBCollectionConfig<
    T extends object = Record<string, unknown>,
    TSchema extends StandardSchemaV1 = never
> = Omit<
    BaseCollectionConfig<T, string, TSchema>,
    `onInsert` | `onUpdate` | `onDelete` | `getKey`
> & {
    /**
     * The RxCollection from a RxDB Database instance.
     */
    rxCollection: RxCollection<T, unknown, unknown, unknown>

    /**
     * The maximum number of documents to read from the RxDB collection
     * in a single batch during the initial sync between RxDB and the
     * in-memory TanStack DB collection.
     *
     * @remarks
     * - Defaults to `1000` if not specified.
     * - Larger values reduce the number of round trips to the storage
     *   engine but increase memory usage per batch.
     * - Smaller values may lower memory usage and allow earlier
     *   streaming of initial results, at the cost of more query calls.
     *
     * Adjust this depending on your expected collection size and
     * performance characteristics of the chosen RxDB storage adapter.
     */
    syncBatchSize?: number
}

/**
 * Creates RxDB collection options for use with a standard Collection
 *
 * @template TExplicit - The explicit type of items in the collection (highest priority)
 * @template TSchema - The schema type for validation and type inference (second priority)
 * @param config - Configuration options for the RxDB collection
 * @returns Collection options with utilities
 */

// Overload for when schema is provided
export function rxdbCollectionOptions<T extends StandardSchemaV1>(
    config: RxDBCollectionConfig<InferSchemaOutput<T>, T>
): CollectionConfig<InferSchemaOutput<T>, string, T> & {
    schema: T
}

// Overload for when no schema is provided
export function rxdbCollectionOptions<T extends object>(
    config: RxDBCollectionConfig<T> & {
        schema?: never // prohibit schema
    }
): CollectionConfig<T, string> & {
    schema?: never // no schema in the result
}

export function rxdbCollectionOptions(config: RxDBCollectionConfig<any, any>) {
    type Row = Record<string, unknown>
    type Key = string // because RxDB primary keys must be strings

    const { ...restConfig } = config
    const rxCollection = config.rxCollection

    // "getKey"
    const primaryPath = rxCollection.schema.primaryPath
    function getKey(item: Record<string, unknown>) {
        const key = item[primaryPath] as string
        return key
    }

    /**
     * "sync"
     * Notice that this describes the Sync between the local RxDB collection
     * and the in-memory tanstack-db collection.
     * It is not about sync between a client and a server!
     */
    type SyncParams = Parameters<SyncConfig<Row, string>[`sync`]>[0]
    const sync: SyncConfig<Row, Key> = {
        sync: (params: SyncParams) => {
            const { begin, write, commit, markReady } = params

            let ready = false
            async function initialFetch() {
                /**
                 * RxDB stores a last-write-time
                 * which can be used to "sort" document writes,
                 * so for initial sync we iterate over that.
                 */
                let cursor: RxDocumentData<Row> | undefined
                const syncBatchSize = config.syncBatchSize
                    ? config.syncBatchSize
                    : 1000
                begin()

                while (!ready) {
                    let query: FilledMangoQuery<Row>
                    if (cursor) {
                        query = {
                            selector: {
                                $or: [
                                    { "_meta.lwt": { $gt: cursor._meta.lwt } },
                                    {
                                        "_meta.lwt": cursor._meta.lwt,
                                        [primaryPath]: {
                                            $gt: getKey(cursor)
                                        }
                                    }
                                ]
                            },
                            sort: [
                                { "_meta.lwt": `asc` },
                                { [primaryPath]: `asc` }
                            ],
                            limit: syncBatchSize,
                            skip: 0
                        }
                    } else {
                        query = {
                            selector: {},
                            sort: [
                                { "_meta.lwt": `asc` },
                                { [primaryPath]: `asc` }
                            ],
                            limit: syncBatchSize,
                            skip: 0
                        }
                    }

                    const preparedQuery = prepareQuery<Row>(
                        rxCollection.storageInstance.schema,
                        query
                    )

                    const docs = await rxCollection
                        .find(preparedQuery.query)
                        .exec()

                    cursor = lastOfArray(docs)
                    if (docs.length === 0) {
                        ready = true
                        break
                    }

                    docs.forEach((d) => {
                        write({
                            type: `insert`,
                            value: stripRxdbFields(clone(d.toJSON()))
                        })
                    })
                }
                commit()
            }

            type WriteMessage = Parameters<typeof write>[0]
            const buffer: Array<WriteMessage> = []
            const queue = (msg: WriteMessage) => {
                if (!ready) {
                    buffer.push(msg)
                    return
                }
                begin()
                write(msg)
                commit()
            }

            let sub: Subscription
            function startOngoingFetch() {
                // Subscribe early and buffer live changes during initial load and ongoing
                sub = rxCollection.$.subscribe((ev) => {
                    const cur: Row = stripRxdbFields(
                        clone(ev.documentData as Row)
                    )
                    switch (ev.operation) {
                        case `INSERT`:
                            queue({ type: `insert`, value: cur })
                            break
                        case `UPDATE`:
                            queue({ type: `update`, value: cur })
                            break
                        case `DELETE`:
                            queue({ type: `delete`, value: cur })
                            break
                    }
                })

                const subs = getFromMapOrCreate(
                    OPEN_RXDB_SUBSCRIPTIONS,
                    rxCollection,
                    () => new Set()
                )
                subs.add(sub)
            }

            async function start() {
                startOngoingFetch()
                await initialFetch()

                if (buffer.length) {
                    begin()
                    for (const msg of buffer) write(msg)
                    commit()
                    buffer.length = 0
                }

                markReady()
            }

            start()

            return () => {
                const subs = getFromMapOrCreate(
                    OPEN_RXDB_SUBSCRIPTIONS,
                    rxCollection,
                    () => new Set()
                )
                subs.delete(sub)
                sub.unsubscribe()
            }
        },
        // Expose the getSyncMetadata function
        getSyncMetadata: undefined
    }

    const collectionConfig: CollectionConfig<Row, string> = {
        ...restConfig,
        getKey: getKey,
        sync,
        onInsert: async (params) => {
            debug(`insert`, params)
            const newItems = params.transaction.mutations.map((m) => m.modified)
            return rxCollection.bulkUpsert(newItems).then((result) => {
                if (result.error.length > 0) {
                    throw rxStorageWriteErrorToRxError(
                        ensureNotFalsy(result.error[0])
                    )
                }
                return result.success
            })
        },
        onUpdate: async (params) => {
            debug(`update`, params)
            const mutations = params.transaction.mutations.filter(
                (m) => m.type === `update`
            )

            for (const mutation of mutations) {
                const newValue = stripRxdbFields(mutation.modified)
                const id = getKey(newValue)
                const doc = await rxCollection.findOne(id).exec()
                if (!doc) {
                    continue
                }
                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) => getKey(mutation.original))
            return rxCollection.bulkRemove(ids).then((result) => {
                if (result.error.length > 0) {
                    throw result.error
                }
                return result.success
            })
        }
    }
    return collectionConfig
}

helper.ts

import { clone } from "rxdb"

const RESERVED_RXDB_FIELDS = ["_rev", "_deleted", "_attachments", "_meta"]

export function stripRxdbFields(obj: Record<string, unknown>) {
    const out = clone(obj)

    for (const k of Object.keys(out)) {
        if (RESERVED_RXDB_FIELDS.includes(k)) {
            delete out[k]
        }
    }

    return out
}

[KyleAMathews]

@pubkey could you take a look at this?

[kevin-dp]

Hi @daveycodez, if you have a branch that improves the types by replacing any types by more concrete types i suggest you open a PR as that would be a great improvement!

[daveycodez]

Sure I can make a PR but I'm hoping to figure out the reasoning for this

/**
 * Instead of doing a RxCollection.query(),
 * we directly query the storage engine of the RxCollection so we do not use the
 * RxCollection document cache because it likely wont be used anyway
 * since most queries will run directly on the tanstack-db side.
 */

Since this by-passes the _deleted checks that RxDB uses, are there any other downsides to porting the adapter to just use RxCollection.query?

[pubkey]

RxCollection.query() uses the RxDocument-cache. This is mostly not helpful if RxDB is used with tanstack/db so this is why we get the plain json data directly out of the storage instance instead.

Adding _deleted: false to the prepared query should fix this.

[daveycodez]

@pubkey I made these changes here:

#667