From 95ae1795877fdf1c24199c151e5a181aecf1e472 Mon Sep 17 00:00:00 2001 From: Jaime Resano Date: Tue, 10 Nov 2026 00:12:08 +0100 Subject: [PATCH 1/2] feat(powersync-db-collection): support numeric collection keys A deserializationSchema may convert `id` to a number, for a backend with integer primary keys. The collection key is now the deserialized row's id, and every path that writes a key derives it from the deserialized row: initial and on-demand loads, the reload's update/insert choice, the tracking-outage sweep, and unload eviction. Local updates and deletes write the key back to SQLite as text. Co-Authored-By: Claude Opus 5.5 --- .changeset/powersync-numeric-keys.md | 5 + docs/collections/powersync-collection.md | 27 +++ .../src/PowerSyncTransactor.ts | 8 +- .../src/definitions.ts | 14 +- .../powersync-db-collection/src/helpers.ts | 15 +- .../powersync-db-collection/src/powersync.ts | 57 +++--- .../tests/numeric-keys.test.ts | 163 ++++++++++++++++++ 7 files changed, 247 insertions(+), 42 deletions(-) create mode 100644 .changeset/powersync-numeric-keys.md create mode 100644 packages/powersync-db-collection/tests/numeric-keys.test.ts diff --git a/.changeset/powersync-numeric-keys.md b/.changeset/powersync-numeric-keys.md new file mode 100644 index 0000000000..fedfe9869b --- /dev/null +++ b/.changeset/powersync-numeric-keys.md @@ -0,0 +1,5 @@ +--- +'@tanstack/powersync-db-collection': patch +--- + +Support numeric collection keys. A `deserializationSchema` may convert `id` to a number, for a backend with integer primary keys; the collection key is then the deserialized number. Sync, on-demand reloads, eviction, and local updates and deletes all use that key. diff --git a/docs/collections/powersync-collection.md b/docs/collections/powersync-collection.md index 472a7e6926..ae6a2aea8b 100644 --- a/docs/collections/powersync-collection.md +++ b/docs/collections/powersync-collection.md @@ -399,6 +399,33 @@ const task = tasksCollection.get("task-1") task.due_date.getTime() // OK - TypeScript knows this is a Date ``` +### Integer Primary Keys + +SQLite stores every PowerSync `id` as text. If your backend uses integer primary keys, convert `id` to a number in `deserializationSchema`. The collection key is then a number, so integer foreign keys match it, for example in `eq(part.id, line.partId)`. + +```typescript +const productSchema = z.object({ + id: z.coerce.number(), // SQLite TEXT '5' to the key 5 + name: z.string().nullable(), +}) + +const products = createCollection( + powerSyncCollectionOptions({ + database: db, + table: APP_SCHEMA.props.products, + schema: productSchema, + deserializationSchema: productSchema, + onDeserializationError: (error) => { + // Present fatal error + }, + }) +) + +products.get(5) +``` + +Updates and deletes write the key back as text. The PowerSync service converts integer `id` columns to text the same way when it syncs them. + ### Optimistic Updates Updates to the collection are applied optimistically to the local state first, then synchronized with PowerSync and the backend. If an error occurs during sync, the changes are automatically rolled back. diff --git a/packages/powersync-db-collection/src/PowerSyncTransactor.ts b/packages/powersync-db-collection/src/PowerSyncTransactor.ts index d78b0a5c54..c1f1f8c628 100644 --- a/packages/powersync-db-collection/src/PowerSyncTransactor.ts +++ b/packages/powersync-db-collection/src/PowerSyncTransactor.ts @@ -2,7 +2,7 @@ import { LogLevels, sanitizeSQL } from '@powersync/common' import { LoadSubsetOperationAbortedError, isFailedStatus } from '@tanstack/db' import DebugModule from 'debug' import { PendingOperationStore } from './PendingOperationStore' -import { asPowerSyncRecord, mapOperationToPowerSync } from './helpers' +import { mapOperationToPowerSync } from './helpers' import type { CommonPowerSyncDatabase, LockContext } from '@powersync/common' import type { PendingMutation, Transaction } from '@tanstack/db' import type { PendingOperation } from './PendingOperationStore' @@ -247,7 +247,7 @@ export class PowerSyncTransactor { SET ${keys.map((key) => `${key} = ?`).join(`, `)} WHERE id = ? `, - [...queryParameters, asPowerSyncRecord(mutation.original).id], + [...queryParameters, String(mutation.key)], ) return @@ -278,14 +278,14 @@ export class PowerSyncTransactor { ` UPDATE ${tableName} SET _deleted = TRUE, _metadata = ? WHERE id = ? `, - [metadataValue, asPowerSyncRecord(mutation.original).id], + [metadataValue, String(mutation.key)], ) } else { await context.execute( ` DELETE FROM ${tableName} WHERE id = ? `, - [asPowerSyncRecord(mutation.original).id], + [String(mutation.key)], ) } }, diff --git a/packages/powersync-db-collection/src/definitions.ts b/packages/powersync-db-collection/src/definitions.ts index b57253a485..fd5f4a2604 100644 --- a/packages/powersync-db-collection/src/definitions.ts +++ b/packages/powersync-db-collection/src/definitions.ts @@ -26,6 +26,16 @@ export type InferPowerSyncOutputType< ? ExtractedTable : InferSchemaOutput +/** + * The collection key: the type of the output row's `id`. It is a string + * unless a schema converts `id`, for example to a number. + */ +export type InferPowerSyncKey = TOutput extends { id: infer TId } + ? TId extends number + ? number + : string + : string + /** * A mapping type for custom serialization of object properties to SQLite-compatible values. * @@ -208,7 +218,7 @@ export type BasePowerSyncCollectionConfig< > = Omit< BaseCollectionConfig< InferPowerSyncOutputType, - string, + InferPowerSyncKey>, TSchema >, `onInsert` | `onUpdate` | `onDelete` | `getKey` | `syncMode` @@ -306,7 +316,7 @@ export type EnhancedPowerSyncCollectionConfig< TSchema extends StandardSchemaV1 = never, > = CollectionConfig< OutputType, - string, + InferPowerSyncKey, TSchema, PowerSyncCollectionUtils > & { diff --git a/packages/powersync-db-collection/src/helpers.ts b/packages/powersync-db-collection/src/helpers.ts index 7bbb8e8b67..8d94823122 100644 --- a/packages/powersync-db-collection/src/helpers.ts +++ b/packages/powersync-db-collection/src/helpers.ts @@ -2,10 +2,12 @@ import { DiffTriggerOperation } from '@powersync/common' import type { ExtractColumnValueType, Table } from '@powersync/common' /** - * All PowerSync table records include a UUID `id` column. + * All PowerSync table records include an `id` column. SQLite stores it as + * text; a deserialization schema may convert it to a number, as for a + * backend with integer primary keys. */ export type PowerSyncRecord = { - id: string + id: string | number [key: string]: unknown } @@ -64,14 +66,7 @@ export type OptionalExtractedTable = OptionalIfUndefined<{ */ export type AnyTableColumnType = { [K in keyof ExtractedTableColumns]: unknown -} & { id: string } - -export function asPowerSyncRecord(record: any): PowerSyncRecord { - if (typeof record.id !== `string`) { - throw new Error(`Record must have a string id field`) - } - return record as PowerSyncRecord -} +} & { id: string | number } /** * Maps {@link DiffTriggerOperation} to TanstackDB operations diff --git a/packages/powersync-db-collection/src/powersync.ts b/packages/powersync-db-collection/src/powersync.ts index dfb52c15f4..dad1044247 100644 --- a/packages/powersync-db-collection/src/powersync.ts +++ b/packages/powersync-db-collection/src/powersync.ts @@ -4,7 +4,7 @@ import { compileSQLite } from './sqlite-compiler' import { PendingOperationStore } from './PendingOperationStore' import { PowerSyncTransactor } from './PowerSyncTransactor' import { DEFAULT_BATCH_SIZE } from './definitions' -import { asPowerSyncRecord, mapOperation } from './helpers' +import { mapOperation } from './helpers' import { convertTableToSchema } from './schema' import { serializeForSQLite } from './serialization' import type { @@ -25,6 +25,7 @@ import type { ConfigWithSQLiteInputType, ConfigWithSQLiteTypes, EnhancedPowerSyncCollectionConfig, + InferPowerSyncKey, InferPowerSyncOutputType, PowerSyncCollectionConfig, PowerSyncCollectionUtils, @@ -267,6 +268,7 @@ function createPowerSyncCollectionConfig< // The collection output type type OutputType = InferPowerSyncOutputType + type PowerSyncKey = InferPowerSyncKey const { viewName, trackMetadata: metadataIsTracked } = table @@ -323,7 +325,7 @@ function createPowerSyncCollectionConfig< * Notice that this describes the Sync between the local SQLite table * and the in-memory tanstack-db collection. */ - const sync: SyncConfig = { + const sync: SyncConfig = { sync: (params) => { const { begin, @@ -428,7 +430,7 @@ function createPowerSyncCollectionConfig< // holds. snapshot?: boolean when: Record - writeType: (rowId: string) => OperationType + writeType: (key: PowerSyncKey) => OperationType batchQuery: ( lockContext: LockContext, batchSize: number, @@ -456,10 +458,10 @@ function createPowerSyncCollectionConfig< const apply = () => { begin({ snapshot }) for (const row of batches.splice(0).flat()) { - write({ - type: writeType(row.id), - value: deserializeSyncRow(row), - }) + // The key comes from the deserialized row: a schema may + // convert SQLite's text id to a number. + const value = deserializeSyncRow(row) + write({ type: writeType(getKey(value)), value }) } appliedReceipts.push(commit()) } @@ -633,7 +635,7 @@ function createPowerSyncCollectionConfig< [DiffTriggerOperation.UPDATE]: `TRUE`, [DiffTriggerOperation.DELETE]: `TRUE`, }, - writeType: (_rowId: string) => `insert`, + writeType: () => `insert`, batchQuery: ( lockContext: LockContext, batchSize: number, @@ -814,22 +816,19 @@ function createPowerSyncCollectionConfig< [DiffTriggerOperation.UPDATE]: `(${newDataWhenClause}) OR (${oldDataWhenClause})`, [DiffTriggerOperation.DELETE]: oldDataWhenClause, }, - writeType: (rowId: string) => - collection.has(rowId) ? `update` : `insert`, + writeType: (key) => { + missing?.delete(key) + return collection.has(key) ? `update` : `insert` + }, batchQuery: ( lockContext: LockContext, batchSize: number, cursor: number, ) => - lockContext - .getAll( - `SELECT * FROM ${viewName} WHERE ${viewWhereClause} LIMIT ? OFFSET ?`, - [batchSize, cursor], - ) - .then((rows) => { - for (const row of rows) missing?.delete(row.id) - return rows - }), + lockContext.getAll( + `SELECT * FROM ${viewName} WHERE ${viewWhereClause} LIMIT ? OFFSET ?`, + [batchSize, cursor], + ), }, appliedReceipts, ) @@ -1013,14 +1012,14 @@ function createPowerSyncCollectionConfig< ): Promise => { const compiledDeparting = compileSQLite({ where: options.where }) const departingWhereSQL = toInlinedWhereClause(compiledDeparting) - let rowsToEvict: Array<{ id: string }> + let rowsToEvict: Array for (;;) { if (stopped) return const revision = trackingRevision const active = activeWhereExpressions() let evictionSQL: string if (active.length === 0) { - evictionSQL = `SELECT id FROM ${viewName} WHERE ${departingWhereSQL}` + evictionSQL = `SELECT * FROM ${viewName} WHERE ${departingWhereSQL}` } else { const combinedRemaining = active.length === 1 @@ -1030,18 +1029,18 @@ function createPowerSyncCollectionConfig< where: combinedRemaining, }) const remainingWhereSQL = toInlinedWhereClause(compiledRemaining) - evictionSQL = `SELECT id FROM ${viewName} WHERE (${departingWhereSQL}) AND NOT (${remainingWhereSQL})` + evictionSQL = `SELECT * FROM ${viewName} WHERE (${departingWhereSQL}) AND NOT (${remainingWhereSQL})` } - rowsToEvict = await database.getAll<{ id: string }>(evictionSQL) + rowsToEvict = await database.getAll(evictionSQL) // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition -- cleanup can run during the query if (stopped) return if (trackingRevision === revision) break } if (rowsToEvict.length > 0) { begin() - for (const { id } of rowsToEvict) { - write({ type: `delete`, key: id }) + for (const row of rowsToEvict) { + write({ type: `delete`, key: getKey(deserializeSyncRow(row)) }) } void commit() } @@ -1182,7 +1181,13 @@ function createPowerSyncCollectionConfig< getSyncMetadata: undefined, } - const getKey = (record: OutputType) => asPowerSyncRecord(record).id + const getKey = (record: OutputType): PowerSyncKey => { + const { id } = record as { id?: unknown } + if (typeof id !== `string` && typeof id !== `number`) { + throw new Error(`Record must have a string or number id field`) + } + return id as PowerSyncKey + } // Keep adapter hooks and serialization options out of the core config. const { diff --git a/packages/powersync-db-collection/tests/numeric-keys.test.ts b/packages/powersync-db-collection/tests/numeric-keys.test.ts new file mode 100644 index 0000000000..9d72bbfce9 --- /dev/null +++ b/packages/powersync-db-collection/tests/numeric-keys.test.ts @@ -0,0 +1,163 @@ +import { randomUUID } from 'node:crypto' +import { tmpdir } from 'node:os' +import { + LogLevels, + PowerSyncDatabase, + Schema, + Table, + column, +} from '@powersync/node' +import { IR, createCollection } from '@tanstack/db' +import { describe, expect, it, onTestFinished, vi } from 'vitest' +import { z } from 'zod' +import { powerSyncCollectionOptions } from '../src' +import type { PowerSyncLogger } from '@powersync/node' + +// SQLite stores every PowerSync id as text. A backend with integer primary +// keys converts `id` to a number in `deserializationSchema`, so the +// collection key is the deserialized number and integer foreign keys match +// it. Every path that writes a key must derive it from the deserialized row, +// not from SQLite's text id. + +const APP_SCHEMA = new Schema({ + products: new Table({ name: column.text, category: column.text }), +}) + +const row = z.object({ + id: z.coerce.number(), + name: z.string().nullable(), + category: z.string().nullable(), +}) + +const categoryEquals = (category: string) => + new IR.Func(`eq`, [ + new IR.PropRef([`category`]), + new IR.Value(category), + ]) + +async function createDatabase() { + const errors: Array = [] + const logger: PowerSyncLogger = { + log({ level, message }) { + if (level >= LogLevels.error) errors.push(message) + }, + } + const db = new PowerSyncDatabase({ + database: { + dbFilename: `test-numeric-keys-${randomUUID()}.sqlite`, + dbLocation: tmpdir(), + implementation: { type: `node:sqlite` }, + }, + schema: APP_SCHEMA, + logger, + }) + onTestFinished(async () => { + await new Promise((resolve) => setTimeout(resolve, 100)) + await db.disconnectAndClear() + await db.close() + }) + await db.disconnectAndClear() + await db.execute( + `INSERT INTO products (id, name, category) VALUES ('1', 'A', 'x'), ('2', 'B', 'x'), ('3', 'C', 'y')`, + ) + return { db, errors } +} + +function createNumericCollection( + db: PowerSyncDatabase, + syncMode: `eager` | `on-demand`, +) { + const collection = createCollection( + powerSyncCollectionOptions({ + database: db, + table: APP_SCHEMA.props.products, + schema: row, + deserializationSchema: row, + onDeserializationError: () => {}, + syncMode, + }), + ) + onTestFinished(() => collection.cleanup()) + return collection +} + +describe(`numeric collection keys`, () => { + it(`syncs and writes rows keyed by a deserialized number`, async () => { + const { db, errors } = await createDatabase() + const collection = createNumericCollection(db, `eager`) + await collection.stateWhenReady() + expect([...collection.keys()].sort()).toEqual([1, 2, 3]) + + await db.execute(`UPDATE products SET name = 'A2' WHERE id = '1'`) + await db.execute(`DELETE FROM products WHERE id = '3'`) + await vi.waitFor(() => { + expect(collection.get(1)?.name).toBe(`A2`) + expect(collection.has(3)).toBe(false) + }) + + await collection + .update(1, (draft) => { + draft.name = `A3` + }) + .when(`settled`) + await collection.delete(2).when(`settled`) + expect(await db.getAll(`SELECT id, name FROM products`)).toEqual([ + { id: `1`, name: `A3` }, + ]) + expect(errors).toEqual([]) + }) + + it(`reloads held rows as updates and evicts them by numeric key`, async () => { + const { db, errors } = await createDatabase() + const collection = createNumericCollection(db, `on-demand`) + const first = { where: categoryEquals(`x`) } + await collection._sync.loadSubset(first) + expect([...collection.keys()].sort()).toEqual([1, 2]) + + // Tracking for the second load re-reads every active predicate, so it + // re-reads the held rows: each must be an update, not an insert. + const second = { where: categoryEquals(`y`) } + await collection._sync.loadSubset(second) + expect([...collection.keys()].sort()).toEqual([1, 2, 3]) + + collection._sync.unloadSubset(second) + await vi.waitFor(() => + expect([...collection.keys()].sort()).toEqual([1, 2]), + ) + collection._sync.unloadSubset(first) + await vi.waitFor(() => expect(collection.size).toBe(0)) + expect(errors).toEqual([]) + }) + + it(`keeps held rows after a tracking outage`, async () => { + vi.useFakeTimers() + const { db } = await createDatabase() + const collection = createNumericCollection(db, `on-demand`) + const first = { where: categoryEquals(`y`) } + const second = { where: categoryEquals(`x`) } + try { + await collection._sync.loadSubset(first) + await collection._sync.loadSubset(second) + const trigger = vi + .spyOn(db.triggers, `createDiffTrigger`) + .mockRejectedValueOnce(new Error(`release rebuild failed`)) + collection._sync.unloadSubset(first) + await vi.waitFor(() => + expect(collection.health.failureReason?.message).toBe( + `release rebuild failed`, + ), + ) + await vi.advanceTimersByTimeAsync(1_000) + await vi.waitFor(() => + expect(trigger.mock.calls.length).toBeGreaterThan(1), + ) + await vi.waitFor(() => expect(collection.health.hasErrors).toBe(false)) + // The rebuild reads the rows still in SQLite; a row it reads is kept. + expect([...collection.keys()].sort()).toEqual([1, 2]) + } finally { + await collection.cleanup() + await vi.runOnlyPendingTimersAsync() + vi.useRealTimers() + } + }) +}) From 2e2728564cb6d69828382b25f158fe01798c7703 Mon Sep 17 00:00:00 2001 From: Jaime Resano Date: Tue, 10 Nov 2026 00:45:15 +0100 Subject: [PATCH 2/2] fix(powersync-db-collection): derive keys before opening a sync transaction Eviction and load batches deserialized each row after begin(). A row that failed deserialization, or had an id that is neither a string nor a number, threw inside the loop and left the sync transaction open; each release retry then opened another. Derive every key first, then open the transaction and write. Co-Authored-By: Claude Opus 5.5 --- .../powersync-db-collection/src/powersync.ts | 29 ++++--- .../tests/numeric-keys.test.ts | 76 +++++++++++++++++++ 2 files changed, 95 insertions(+), 10 deletions(-) diff --git a/packages/powersync-db-collection/src/powersync.ts b/packages/powersync-db-collection/src/powersync.ts index dad1044247..8808f3d3e0 100644 --- a/packages/powersync-db-collection/src/powersync.ts +++ b/packages/powersync-db-collection/src/powersync.ts @@ -452,16 +452,22 @@ function createPowerSyncCollectionConfig< // A load of every row is one snapshot transaction: during a // reload it replaces the kept rows, so a row the table no // longer has goes. A subset load commits each batch. A - // transaction begins only once its rows are read, so a batch - // that fails leaves none open. + // transaction begins only once its rows are read and + // deserialized, so a batch that fails leaves none open. const batches: Array> = [] const apply = () => { + // The key comes from the deserialized row: a schema may + // convert SQLite's text id to a number. + const entries = batches + .splice(0) + .flat() + .map((row) => { + const value = deserializeSyncRow(row) + return [getKey(value), value] as const + }) begin({ snapshot }) - for (const row of batches.splice(0).flat()) { - // The key comes from the deserialized row: a schema may - // convert SQLite's text id to a number. - const value = deserializeSyncRow(row) - write({ type: writeType(getKey(value)), value }) + for (const [key, value] of entries) { + write({ type: writeType(key), value }) } appliedReceipts.push(commit()) } @@ -1038,10 +1044,13 @@ function createPowerSyncCollectionConfig< if (trackingRevision === revision) break } if (rowsToEvict.length > 0) { + // Derive every key before the transaction opens: a row that fails + // deserialization must not leave it open. + const keys = rowsToEvict.map((row) => + getKey(deserializeSyncRow(row)), + ) begin() - for (const row of rowsToEvict) { - write({ type: `delete`, key: getKey(deserializeSyncRow(row)) }) - } + for (const key of keys) write({ type: `delete`, key }) void commit() } await rebuildTracking() diff --git a/packages/powersync-db-collection/tests/numeric-keys.test.ts b/packages/powersync-db-collection/tests/numeric-keys.test.ts index 9d72bbfce9..0b971081a9 100644 --- a/packages/powersync-db-collection/tests/numeric-keys.test.ts +++ b/packages/powersync-db-collection/tests/numeric-keys.test.ts @@ -161,3 +161,79 @@ describe(`numeric collection keys`, () => { } }) }) + +// Deriving a key deserializes the row, and deserialization can fail. A sync +// transaction opens only after every key is derived, so a failing row leaves +// no open transaction behind. +describe(`a row that fails deserialization`, () => { + function createGatedCollection(db: PowerSyncDatabase) { + const gate = { valid: true } + const gated = row.extend({ + id: z.coerce.number().refine(() => gate.valid), + }) + const collection = createCollection( + powerSyncCollectionOptions({ + database: db, + table: APP_SCHEMA.props.products, + schema: gated, + deserializationSchema: gated, + onDeserializationError: () => {}, + syncMode: `on-demand`, + }), + ) + const openTransactions = () => + collection._state.pendingSyncedTransactions.filter( + (transaction) => !transaction.committed, + ).length + return { collection, gate, openTransactions } + } + + it(`fails a load without leaving a sync transaction open`, async () => { + const { db } = await createDatabase() + const { collection, gate, openTransactions } = createGatedCollection(db) + try { + gate.valid = false + await expect( + Promise.resolve( + collection._sync.loadSubset({ where: categoryEquals(`x`) }), + ), + ).rejects.toThrow() + expect(openTransactions()).toBe(0) + expect(collection.size).toBe(0) + + gate.valid = true + await collection._sync.loadSubset({ where: categoryEquals(`x`) }) + expect([...collection.keys()].sort()).toEqual([1, 2]) + } finally { + await collection.cleanup() + } + }) + + it(`retries an eviction without leaving a sync transaction open`, async () => { + vi.useFakeTimers() + const { db, errors } = await createDatabase() + const { collection, gate, openTransactions } = createGatedCollection(db) + const first = { where: categoryEquals(`x`) } + try { + await collection._sync.loadSubset(first) + gate.valid = false + collection._sync.unloadSubset(first) + await vi.waitFor(() => + expect(errors).toContainEqual( + expect.stringContaining(`Could not release subset tracking`), + ), + ) + expect(openTransactions()).toBe(0) + expect([...collection.keys()].sort()).toEqual([1, 2]) + + gate.valid = true + await vi.advanceTimersByTimeAsync(1_000) + await vi.waitFor(() => expect(collection.size).toBe(0)) + expect(openTransactions()).toBe(0) + } finally { + await collection.cleanup() + await vi.runOnlyPendingTimersAsync() + vi.useRealTimers() + } + }) +})