From 3124c9e49e831a0a012295d4e82a55cfda95f848 Mon Sep 17 00:00:00 2001 From: Christiaan Landman Date: Fri, 12 Jun 2026 16:01:44 +0200 Subject: [PATCH 01/13] Added attachments support. --- packages/powersync-db-collection/package.json | 4 +- .../src/attachments.ts | 181 ++++++++++++++++++ pnpm-lock.yaml | 23 ++- 3 files changed, 197 insertions(+), 11 deletions(-) create mode 100644 packages/powersync-db-collection/src/attachments.ts diff --git a/packages/powersync-db-collection/package.json b/packages/powersync-db-collection/package.json index 93444c4f26..66f2af4ae5 100644 --- a/packages/powersync-db-collection/package.json +++ b/packages/powersync-db-collection/package.json @@ -59,10 +59,10 @@ "p-defer": "^4.0.1" }, "peerDependencies": { - "@powersync/common": "^1.41.0" + "@powersync/common": "^1.54.0" }, "devDependencies": { - "@powersync/common": "1.49.0", + "@powersync/common": "1.54.0", "@powersync/node": "0.18.1", "@types/debug": "^4.1.12", "@vitest/coverage-istanbul": "^3.2.4", diff --git a/packages/powersync-db-collection/src/attachments.ts b/packages/powersync-db-collection/src/attachments.ts new file mode 100644 index 0000000000..9359abfc9e --- /dev/null +++ b/packages/powersync-db-collection/src/attachments.ts @@ -0,0 +1,181 @@ +import { + AttachmentQueue, + AttachmentState, + AttachmentTable, + Schema, +} from '@powersync/common' +import { createTransaction } from '@tanstack/db' +import { PowerSyncTransactor } from './PowerSyncTransactor' + +import type { + AbstractPowerSyncDatabase, + AttachmentData, + AttachmentErrorHandler, + ILogger, + LocalStorageAdapter, + RemoteStorageAdapter, + WatchedAttachmentItem, +} from '@powersync/common' +import type { Collection } from '@tanstack/db' + +type AttachmentQueueRow = (typeof _tmpSchema)['types']['attachments'] + +/** + * This extends the default AttachmentQueue constructor params + * FIXME(powersync) we should export this type from the common SDK. + */ +type TanStackDBAttachmentQueueOptions = { + db: AbstractPowerSyncDatabase + /** + * For TanStack, we want access to the synced TanStackDB collection. + * In order to have the same relational data be set in a single transaction. + * This also allows for joining both TanStackDB collections. + */ + attachmentsCollection: Collection + remoteStorage: RemoteStorageAdapter + localStorage: LocalStorageAdapter + watchAttachments: ( + onUpdate: (attachment: Array) => Promise, + signal: AbortSignal, + ) => void + tableName?: string + logger?: ILogger + syncIntervalMs?: number + syncThrottleDuration?: number + downloadAttachments?: boolean + archivedCacheLimit?: number + errorHandler?: AttachmentErrorHandler +} + +interface SaveFileTanStackOptions { + data: AttachmentData + fileExtension: string + mediaType?: string + metaData?: string + id?: string + /** + * Note that this is called inside a synchronous TanStackDB transaction, + * any mutations made to other collections will be in the same transaction. + */ + updateHook?: (attachment: AttachmentQueueRow) => Promise +} + +interface DeleteFileTanStackOptions { + id: string + updateHook?: (attachment: AttachmentQueueRow) => Promise +} + +const _tmpSchema = new Schema({ + attachments: new AttachmentTable(), +}) + +/** + * A custom extension of the PowerSyncAttachmentQueue for TanStackDB. + */ +export class TanStackDBAttachmentQueue extends AttachmentQueue { + readonly powersync: AbstractPowerSyncDatabase + readonly collection: Collection + + constructor(params: TanStackDBAttachmentQueueOptions) { + super(params) + this.powersync = params.db + this.collection = params.attachmentsCollection + } + + /** + * Saves a file to local storage and queues it for upload to remote storage. + * + * Exposes an `updateHook` option which is called inside a TanStackDB transaction, + * relational associations with the provided attachment ID should be made in this hook. + */ + async saveFileTanStack({ + data, + fileExtension, + mediaType, + metaData, + id, + updateHook, + }: SaveFileTanStackOptions): Promise { + const resolvedId = id ?? (await this.generateAttachmentId()) + const filename = `${resolvedId}.${fileExtension}` + const localUri = this.localStorage.getLocalUri(filename) + const size = await this.localStorage.saveFile(localUri, data) + + const attachment: AttachmentQueueRow = { + id: resolvedId, + filename, + media_type: mediaType ?? null, + local_uri: localUri, + state: AttachmentState.QUEUED_UPLOAD, + has_synced: 0, + size, + timestamp: new Date().getTime(), + meta_data: metaData ?? null, + } + + /** + * We use the attachmentService lock to prevent attachment queue race conditions — specifically, + * it stops the watcher from treating a newly inserted attachment record as one that needs + * to be downloaded. + * */ + await this.withAttachmentContext(async (ctx) => { + const tanStackDBTransaction = createTransaction({ + autoCommit: false, + mutationFn: async ({ transaction }) => { + await new PowerSyncTransactor({ + database: ctx.db, + }).applyTransaction(transaction) + }, + }) + + tanStackDBTransaction.mutate(() => { + this.collection.insert(attachment) + // allow the user to associate values in this transaction + updateHook?.(attachment) + }) + + await tanStackDBTransaction.commit() + }) + + return attachment + } + + /** + * Queues a file for deletion from local and remote storage. + * + * Exposes an `updateHook` option which is called inside a TanStackDB transaction, + * relational associations with the provided attachment ID should be cleaned up in this hook. + */ + async deleteFileTanStack({ + id, + updateHook, + }: DeleteFileTanStackOptions): Promise { + await this.withAttachmentContext(async (ctx) => { + const tanStackDBTransaction = createTransaction({ + autoCommit: false, + mutationFn: async ({ transaction }) => { + await new PowerSyncTransactor({ + database: ctx.db, + }).applyTransaction(transaction) + }, + }) + + tanStackDBTransaction.mutate(() => { + const attachment = this.collection.get(id) + if (!attachment) { + throw new Error(`Attachment with id ${id} not found`) + } + + this.collection.update(id, (draft) => { + draft.state = AttachmentState.QUEUED_DELETE + draft.has_synced = 0 + }) + + // allow the user to associate values in this transaction + updateHook?.(attachment) + }) + + await tanStackDBTransaction.commit() + }) + } +} diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index ad174b46b5..c4562916df 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1338,11 +1338,11 @@ importers: version: 4.0.1 devDependencies: '@powersync/common': - specifier: 1.49.0 - version: 1.49.0 + specifier: 1.54.0 + version: 1.54.0 '@powersync/node': specifier: 0.18.1 - version: 0.18.1(@powersync/common@1.49.0)(better-sqlite3@12.8.0) + version: 0.18.1(@powersync/common@1.54.0)(better-sqlite3@12.8.0) '@types/debug': specifier: ^4.1.12 version: 4.1.12 @@ -4831,8 +4831,8 @@ packages: '@poppinss/exception@1.2.3': resolution: {integrity: sha512-dCED+QRChTVatE9ibtoaxc+WkdzOSjYTKi/+uacHWIsfodVfpsueo3+DKpgU5Px8qXjgmXkSvhXvSCz3fnP9lw==} - '@powersync/common@1.49.0': - resolution: {integrity: sha512-g6uonubvtmtyx8hS/G5trg9LsBvzHY3tAKHiV7SIQV3Xyz9ONM6NNnjDMP2vcLZVmsOSi8x/QJZmy/ig1YtBMg==} + '@powersync/common@1.54.0': + resolution: {integrity: sha512-/gzitw4iQL4UI7ILf7TUzCy/cfbDJGU3/aiN/ciaLtDd2Uts3wYARVKclSW0OJhPPisKCX0E8Ev/iZGQPbTgDA==} '@powersync/node@0.18.1': resolution: {integrity: sha512-fcTICgs61CAEb39xiC7pedYsPgbjUInJ/47dr7RIdnEHpAgjWH8bW95/b70qK1fQUANy9lKBBF3PcmfswVgfCw==} @@ -9372,6 +9372,9 @@ packages: js-base64@3.7.8: resolution: {integrity: sha512-hNngCeKxIUQiEUN3GPJOkz4wF/YvdUdbNL9hsBcMQTkKzboD7T/q3OYOuuPZLUE6dBxSGpwhk5mwuDud7JVAow==} + js-logger@1.6.1: + resolution: {integrity: sha512-yTgMCPXVjhmg28CuUH8CKjU+cIKL/G+zTu4Fn4lQxs8mRFH/03QTNvEFngcxfg/gRDiQAOoyCKmMTOm9ayOzXA==} + js-tokens@4.0.0: resolution: {integrity: sha512-RdJUflcE3cUzKiMqQgsCu06FPu9UdIJO0beYbPhHN4k6apgJtifcoCtT9bcxOpYBtpD2kCM6Sbzg4CausW/PKQ==} @@ -16480,14 +16483,14 @@ snapshots: '@poppinss/exception@1.2.3': {} - '@powersync/common@1.49.0': + '@powersync/common@1.54.0': dependencies: - async-mutex: 0.5.0 event-iterator: 2.0.0 + js-logger: 1.6.1 - '@powersync/node@0.18.1(@powersync/common@1.49.0)(better-sqlite3@12.8.0)': + '@powersync/node@0.18.1(@powersync/common@1.54.0)(better-sqlite3@12.8.0)': dependencies: - '@powersync/common': 1.49.0 + '@powersync/common': 1.54.0 async-mutex: 0.5.0 bson: 6.10.4 comlink: 4.4.2 @@ -22133,6 +22136,8 @@ snapshots: js-base64@3.7.8: {} + js-logger@1.6.1: {} + js-tokens@4.0.0: {} js-tokens@9.0.1: {} From d18ab196cd1b1c320b990e47fde5c8044d82082d Mon Sep 17 00:00:00 2001 From: Christiaan Landman Date: Thu, 18 Jun 2026 16:13:35 +0200 Subject: [PATCH 02/13] Types and tests. --- packages/powersync-db-collection/package.json | 4 +- .../src/attachments.ts | 36 +- packages/powersync-db-collection/src/index.ts | 1 + .../tests/attachments.test.ts | 401 ++++++++++++++++++ pnpm-lock.yaml | 22 +- 5 files changed, 419 insertions(+), 45 deletions(-) create mode 100644 packages/powersync-db-collection/tests/attachments.test.ts diff --git a/packages/powersync-db-collection/package.json b/packages/powersync-db-collection/package.json index 66f2af4ae5..9374371824 100644 --- a/packages/powersync-db-collection/package.json +++ b/packages/powersync-db-collection/package.json @@ -59,10 +59,10 @@ "p-defer": "^4.0.1" }, "peerDependencies": { - "@powersync/common": "^1.54.0" + "@powersync/common": "^1.55.0" }, "devDependencies": { - "@powersync/common": "1.54.0", + "@powersync/common": "1.55.0", "@powersync/node": "0.18.1", "@types/debug": "^4.1.12", "@vitest/coverage-istanbul": "^3.2.4", diff --git a/packages/powersync-db-collection/src/attachments.ts b/packages/powersync-db-collection/src/attachments.ts index 9359abfc9e..c3c9d8f677 100644 --- a/packages/powersync-db-collection/src/attachments.ts +++ b/packages/powersync-db-collection/src/attachments.ts @@ -10,44 +10,22 @@ import { PowerSyncTransactor } from './PowerSyncTransactor' import type { AbstractPowerSyncDatabase, AttachmentData, - AttachmentErrorHandler, - ILogger, - LocalStorageAdapter, - RemoteStorageAdapter, - WatchedAttachmentItem, + AttachmentQueueOptions, } from '@powersync/common' import type { Collection } from '@tanstack/db' -type AttachmentQueueRow = (typeof _tmpSchema)['types']['attachments'] +export type AttachmentQueueRow = (typeof _tmpSchema)['types']['attachments'] -/** - * This extends the default AttachmentQueue constructor params - * FIXME(powersync) we should export this type from the common SDK. - */ -type TanStackDBAttachmentQueueOptions = { - db: AbstractPowerSyncDatabase +export type TanStackDBAttachmentQueueOptions = AttachmentQueueOptions & { /** * For TanStack, we want access to the synced TanStackDB collection. * In order to have the same relational data be set in a single transaction. * This also allows for joining both TanStackDB collections. */ - attachmentsCollection: Collection - remoteStorage: RemoteStorageAdapter - localStorage: LocalStorageAdapter - watchAttachments: ( - onUpdate: (attachment: Array) => Promise, - signal: AbortSignal, - ) => void - tableName?: string - logger?: ILogger - syncIntervalMs?: number - syncThrottleDuration?: number - downloadAttachments?: boolean - archivedCacheLimit?: number - errorHandler?: AttachmentErrorHandler + attachmentsCollection: Collection } -interface SaveFileTanStackOptions { +export interface SaveFileTanStackOptions { data: AttachmentData fileExtension: string mediaType?: string @@ -60,7 +38,7 @@ interface SaveFileTanStackOptions { updateHook?: (attachment: AttachmentQueueRow) => Promise } -interface DeleteFileTanStackOptions { +export interface DeleteFileTanStackOptions { id: string updateHook?: (attachment: AttachmentQueueRow) => Promise } @@ -74,7 +52,7 @@ const _tmpSchema = new Schema({ */ export class TanStackDBAttachmentQueue extends AttachmentQueue { readonly powersync: AbstractPowerSyncDatabase - readonly collection: Collection + readonly collection: Collection constructor(params: TanStackDBAttachmentQueueOptions) { super(params) diff --git a/packages/powersync-db-collection/src/index.ts b/packages/powersync-db-collection/src/index.ts index f8d0928056..f96a7a0ee4 100644 --- a/packages/powersync-db-collection/src/index.ts +++ b/packages/powersync-db-collection/src/index.ts @@ -1,3 +1,4 @@ +export * from './attachments' export * from './definitions' export * from './powersync' export * from './PowerSyncTransactor' diff --git a/packages/powersync-db-collection/tests/attachments.test.ts b/packages/powersync-db-collection/tests/attachments.test.ts new file mode 100644 index 0000000000..411c7cd753 --- /dev/null +++ b/packages/powersync-db-collection/tests/attachments.test.ts @@ -0,0 +1,401 @@ +import { randomUUID } from 'node:crypto' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { + AttachmentState, + AttachmentTable, + Schema, + Table, + column, +} from '@powersync/common' +import { NodeFileSystemAdapter, PowerSyncDatabase } from '@powersync/node' +import { + createCollection, + isNull, + liveQueryCollectionOptions, + not, +} from '@tanstack/db' +import { describe, expect, it, onTestFinished, vi } from 'vitest' +import { powerSyncCollectionOptions } from '../src' +import { TanStackDBAttachmentQueue } from '../src/attachments' +import { TEST_DATABASE_IMPLEMENTATION } from './test-db-implementation' +import type { + AttachmentErrorHandler, + RemoteStorageAdapter, + WatchedAttachmentItem, +} from '@powersync/common' + +// A minimal valid 1x1 pixel JPEG used as the remote payload for downloads. +const MOCK_JPEG_U8A = [ + 0xff, 0xd8, 0xff, 0xe0, 0x00, 0x10, 0x4a, 0x46, 0x49, 0x46, 0x00, 0x01, 0x01, + 0x01, 0x00, 0x00, 0x01, 0x00, 0x01, 0x00, 0x00, 0xff, 0xd9, +] +const createMockJpegBuffer = (): ArrayBuffer => + new Uint8Array(MOCK_JPEG_U8A).buffer + +const SYNC_INTERVAL_MS = 300 +const WAIT_TIMEOUT = 8000 + +const APP_SCHEMA = new Schema({ + users: new Table({ + name: column.text, + email: column.text, + photo_id: column.text, + }), + attachments: new AttachmentTable(), +}) + +type WatchAttachments = ( + onUpdate: (attachments: Array) => Promise, + signal: AbortSignal, +) => void + +const describePowerSync = TEST_DATABASE_IMPLEMENTATION + ? describe + : describe.skip + +describePowerSync(`PowerSync AttachmentQueue (TanStackDB)`, () => { + async function setup() { + const db = new PowerSyncDatabase({ + database: { + dbFilename: `attachments-test-${randomUUID()}.sqlite`, + dbLocation: tmpdir(), + implementation: TEST_DATABASE_IMPLEMENTATION, + }, + schema: APP_SCHEMA, + }) + await db.disconnectAndClear() + + const localStorage = new NodeFileSystemAdapter( + join(tmpdir(), `ps-attachments-${randomUUID()}`), + ) + await localStorage.initialize() + + const uploadFile = vi.fn(() => + Promise.resolve(), + ) + const downloadFile = vi.fn(() => + Promise.resolve(createMockJpegBuffer()), + ) + const deleteFile = vi.fn(() => + Promise.resolve(), + ) + const remoteStorage: RemoteStorageAdapter = { + uploadFile, + downloadFile, + deleteFile, + } + + const attachmentsCollection = createCollection( + powerSyncCollectionOptions({ + database: db, + table: APP_SCHEMA.props.attachments, + }), + ) + const usersCollection = createCollection( + powerSyncCollectionOptions({ + database: db, + table: APP_SCHEMA.props.users, + }), + ) + await Promise.all([ + attachmentsCollection.stateWhenReady(), + usersCollection.stateWhenReady(), + ]) + + onTestFinished(async () => { + attachmentsCollection.cleanup() + usersCollection.cleanup() + await db.disconnectAndClear() + await db.close() + await localStorage.clear().catch(() => {}) + }) + + function createQueue( + overrides: { + watchAttachments?: WatchAttachments + archivedCacheLimit?: number + errorHandler?: AttachmentErrorHandler + remoteStorage?: RemoteStorageAdapter + } = {}, + ) { + const queue = new TanStackDBAttachmentQueue({ + db, + attachmentsCollection, + remoteStorage: overrides.remoteStorage ?? remoteStorage, + localStorage, + watchAttachments: overrides.watchAttachments ?? watchPhotoIds, + syncIntervalMs: SYNC_INTERVAL_MS, + archivedCacheLimit: overrides.archivedCacheLimit ?? 0, + errorHandler: overrides.errorHandler, + }) + onTestFinished(() => queue.stopSync()) + return queue + } + + // Reports every photo_id referenced by the users collection as a watched + // attachment. This mirrors how an application links its domain model to the + // attachment queue using a TanStack DB live query rather than a raw SQL + // watch: the `photo_id IS NOT NULL` filter lives in the query, and each + // change re-emits the full set of referenced ids. + const watchPhotoIdsWith = ( + toItem: (photoId: string) => WatchedAttachmentItem, + ): WatchAttachments => { + return async (onUpdate, signal) => { + const livePhotoIds = createCollection( + liveQueryCollectionOptions({ + query: (q) => + q + .from({ user: usersCollection }) + .where(({ user }) => not(isNull(user.photo_id))) + .select(({ user }) => ({ photo_id: user.photo_id })), + }), + ) + + const emit = () => + void onUpdate( + livePhotoIds.toArray + .map((row) => row.photo_id) + .filter((photoId): photoId is string => photoId != null) + .map(toItem), + ) + + // Emit the current snapshot once ready, then on every change. + await livePhotoIds.stateWhenReady() + emit() + const subscription = livePhotoIds.subscribeChanges(() => emit()) + + signal.addEventListener(`abort`, () => { + subscription.unsubscribe() + livePhotoIds.cleanup() + }) + } + } + + const watchPhotoIds = watchPhotoIdsWith((id) => ({ + id, + fileExtension: `jpg`, + })) + + return { + db, + localStorage, + remoteStorage, + uploadFile, + downloadFile, + deleteFile, + attachmentsCollection, + usersCollection, + createQueue, + watchPhotoIds, + watchPhotoIdsWith, + } + } + + /** Waits until the attachment with `id` reaches the expected state. */ + function waitForState( + collection: { get: (id: string) => TRow | undefined }, + id: string, + state: AttachmentState, + ): Promise { + return vi.waitFor( + () => { + const attachment = collection.get(id) + expect( + (attachment as { state?: AttachmentState } | undefined)?.state, + ).toBe(state) + return attachment! + }, + { timeout: WAIT_TIMEOUT, interval: 50 }, + ) + } + + describe(`saveFileTanStack`, () => { + it(`writes the local file and inserts a QUEUED_UPLOAD row into the collection`, async () => { + const { createQueue, attachmentsCollection, localStorage } = await setup() + const queue = createQueue() + + const data = new Uint8Array(123).fill(42).buffer + const record = await queue.saveFileTanStack({ + data, + fileExtension: `jpg`, + mediaType: `image/jpeg`, + }) + + expect(record.size).toBe(123) + expect(record.state).toBe(AttachmentState.QUEUED_UPLOAD) + expect(record.media_type).toBe(`image/jpeg`) + expect(record.filename).toBe(`${record.id}.jpg`) + expect(record.has_synced).toBe(0) + + // The file should exist on disk at the returned local_uri. + expect(await localStorage.fileExists(record.local_uri)).toBe(true) + + // The row should be reflected in the collection once it syncs back. + await waitForState( + attachmentsCollection, + record.id, + AttachmentState.QUEUED_UPLOAD, + ) + }) + + it(`commits the updateHook mutation atomically with the attachment row`, async () => { + const { createQueue, attachmentsCollection, usersCollection } = + await setup() + const queue = createQueue() + + const userId = randomUUID() + const record = await queue.saveFileTanStack({ + data: createMockJpegBuffer(), + fileExtension: `jpg`, + updateHook: async (attachment) => { + usersCollection.insert({ + id: userId, + name: `steven`, + email: `steven@journeyapps.com`, + photo_id: attachment.id, + }) + }, + }) + + // Both the attachment and the linked user row should appear together. + await waitForState( + attachmentsCollection, + record.id, + AttachmentState.QUEUED_UPLOAD, + ) + await vi.waitFor( + () => { + const user = usersCollection.get(userId) + expect(user?.photo_id).toBe(record.id) + }, + { timeout: WAIT_TIMEOUT, interval: 50 }, + ) + }) + + it(`uploads the saved file and transitions it to SYNCED`, async () => { + const { + createQueue, + attachmentsCollection, + usersCollection, + uploadFile, + } = await setup() + const queue = createQueue() + await queue.startSync() + + const userId = randomUUID() + const record = await queue.saveFileTanStack({ + data: createMockJpegBuffer(), + fileExtension: `jpg`, + updateHook: async (attachment) => { + usersCollection.insert({ + id: userId, + name: `steven`, + email: `steven@journeyapps.com`, + photo_id: attachment.id, + }) + }, + }) + + await waitForState( + attachmentsCollection, + record.id, + AttachmentState.SYNCED, + ) + + expect(uploadFile).toHaveBeenCalled() + const [, uploadedAttachment] = uploadFile.mock.calls[0]! + expect(uploadedAttachment.id).toBe(record.id) + }) + + it(`honours a caller-supplied id`, async () => { + const { createQueue } = await setup() + const queue = createQueue() + + const id = `my-custom-id` + const record = await queue.saveFileTanStack({ + id, + data: createMockJpegBuffer(), + fileExtension: `png`, + }) + + expect(record.id).toBe(id) + expect(record.filename).toBe(`${id}.png`) + }) + }) + + describe(`deleteFileTanStack`, () => { + it(`queues an existing attachment for deletion and removes the local file`, async () => { + const { + createQueue, + attachmentsCollection, + usersCollection, + localStorage, + } = await setup() + const queue = createQueue() + await queue.startSync() + + const userId = randomUUID() + const record = await queue.saveFileTanStack({ + data: createMockJpegBuffer(), + fileExtension: `jpg`, + updateHook: async (attachment) => { + usersCollection.insert({ + id: userId, + name: `steven`, + email: `steven@journeyapps.com`, + photo_id: attachment.id, + }) + }, + }) + + await waitForState( + attachmentsCollection, + record.id, + AttachmentState.SYNCED, + ) + + await queue.deleteFileTanStack({ + id: record.id, + updateHook: async (attachment) => { + usersCollection.update(userId, (draft) => { + if (draft.photo_id === attachment.id) { + draft.photo_id = null + } + }) + }, + }) + + // It should immediately be marked for deletion (and no longer synced). + const queued = attachmentsCollection.get(record.id) + expect(queued?.state).toBe(AttachmentState.QUEUED_DELETE) + expect(queued?.has_synced).toBe(0) + + // The user reference should have been cleared in the same transaction. + expect(usersCollection.get(userId)?.photo_id).toBeNull() + + // Eventually the row and the local file are removed. + await vi.waitFor( + () => expect(attachmentsCollection.get(record.id)).toBeUndefined(), + { timeout: WAIT_TIMEOUT, interval: 50 }, + ) + expect(await localStorage.fileExists(record.local_uri)).toBe(false) + }) + + it(`throws for an unknown id and commits nothing`, async () => { + const { createQueue, attachmentsCollection, usersCollection } = + await setup() + const queue = createQueue() + + const hook = vi.fn() + await expect( + queue.deleteFileTanStack({ id: `does-not-exist`, updateHook: hook }), + ).rejects.toThrow(/not found/i) + + // The failing transaction must not have run the hook or touched state. + expect(hook).not.toHaveBeenCalled() + expect(attachmentsCollection.get(`does-not-exist`)).toBeUndefined() + expect(usersCollection.size).toBe(0) + }) + }) +}) diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index c4562916df..3852ece3ed 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1338,11 +1338,11 @@ importers: version: 4.0.1 devDependencies: '@powersync/common': - specifier: 1.54.0 - version: 1.54.0 + specifier: 1.55.0 + version: 1.55.0 '@powersync/node': specifier: 0.18.1 - version: 0.18.1(@powersync/common@1.54.0)(better-sqlite3@12.8.0) + version: 0.18.1(@powersync/common@1.55.0)(better-sqlite3@12.8.0) '@types/debug': specifier: ^4.1.12 version: 4.1.12 @@ -4831,8 +4831,8 @@ packages: '@poppinss/exception@1.2.3': resolution: {integrity: sha512-dCED+QRChTVatE9ibtoaxc+WkdzOSjYTKi/+uacHWIsfodVfpsueo3+DKpgU5Px8qXjgmXkSvhXvSCz3fnP9lw==} - '@powersync/common@1.54.0': - resolution: {integrity: sha512-/gzitw4iQL4UI7ILf7TUzCy/cfbDJGU3/aiN/ciaLtDd2Uts3wYARVKclSW0OJhPPisKCX0E8Ev/iZGQPbTgDA==} + '@powersync/common@1.55.0': + resolution: {integrity: sha512-c9K2Gac9wOB4ijVnQT388g7Yeuh6VrfJZFXaL6Rag9Fp7F8JzAYcrWaULLTdMnzm7VW6b6XT5M2ip5yLS8oSBg==} '@powersync/node@0.18.1': resolution: {integrity: sha512-fcTICgs61CAEb39xiC7pedYsPgbjUInJ/47dr7RIdnEHpAgjWH8bW95/b70qK1fQUANy9lKBBF3PcmfswVgfCw==} @@ -8173,9 +8173,6 @@ packages: resolution: {integrity: sha512-aIL5Fx7mawVa300al2BnEE4iNvo1qETxLrPI/o05L7z6go7fCw1J6EQmbK4FmJ2AS7kgVF/KEZWufBfdClMcPg==} engines: {node: '>= 0.6'} - event-iterator@2.0.0: - resolution: {integrity: sha512-KGft0ldl31BZVV//jj+IAIGCxkvvUkkON+ScH6zfoX+l+omX6001ggyRSpI0Io2Hlro0ThXotswCtfzS8UkIiQ==} - event-reduce-js@5.2.7: resolution: {integrity: sha512-Vi6aIiAmakzx81JAwhw8L988aSX5a3ZqqVjHyZa9xFU6P4oT1IotoDreWtjNlS+fvEnASvyIQT565nmkOtns/Q==} engines: {node: '>=16'} @@ -16483,14 +16480,13 @@ snapshots: '@poppinss/exception@1.2.3': {} - '@powersync/common@1.54.0': + '@powersync/common@1.55.0': dependencies: - event-iterator: 2.0.0 js-logger: 1.6.1 - '@powersync/node@0.18.1(@powersync/common@1.54.0)(better-sqlite3@12.8.0)': + '@powersync/node@0.18.1(@powersync/common@1.55.0)(better-sqlite3@12.8.0)': dependencies: - '@powersync/common': 1.54.0 + '@powersync/common': 1.55.0 async-mutex: 0.5.0 bson: 6.10.4 comlink: 4.4.2 @@ -20580,8 +20576,6 @@ snapshots: etag@1.8.1: {} - event-iterator@2.0.0: {} - event-reduce-js@5.2.7: dependencies: array-push-at-sort-position: 4.0.1 From 91c24c6df4579bcca320227a9da4e222ccc42215 Mon Sep 17 00:00:00 2001 From: Christiaan Landman Date: Tue, 23 Jun 2026 13:49:32 +0200 Subject: [PATCH 03/13] Docs. --- docs/collections/powersync-collection.md | 167 ++++++++++++++++++ .../src/attachments.ts | 4 + 2 files changed, 171 insertions(+) diff --git a/docs/collections/powersync-collection.md b/docs/collections/powersync-collection.md index c8ddbabbbe..b484c5bb92 100644 --- a/docs/collections/powersync-collection.md +++ b/docs/collections/powersync-collection.md @@ -1099,4 +1099,171 @@ const liveQuery = createLiveQueryCollection({ completed: todo.completed, })), }) +``` + +## Attachments + +`@tanstack/powersync-db-collection` ships `TanStackDBAttachmentQueue`, an [`AttachmentQueue`](https://docs.powersync.com/usage/use-case-examples/attachments-files) whose file operations commit inside a TanStack DB collection transaction. This lets you create (or delete) an attachment and mutate a related collection row (for example, setting `lists.photo_id`) atomically in a single transaction, instead of issuing two independent writes. + +The queue extends PowerSync's `AttachmentQueue`, so the generic concepts are unchanged and documented once in the SDK. + +> This section only covers what is specific to the TanStack DB integration. For storage adapters (local and remote), the `AttachmentTable` schema primitive, error-handling/retry semantics, and the `startSync()` / `stopSync()` lifecycle, see the [PowerSync attachments documentation](https://docs.powersync.com/usage/use-case-examples/attachments-files). + +### Prerequisites + +These are standard PowerSync attachment requirements. See the SDK attachments docs for details. + +- An `AttachmentTable` in your schema: + + ```ts + import { AttachmentTable, Schema } from "@powersync/web" + + const APP_SCHEMA = new Schema({ + // ...your tables + attachments: new AttachmentTable(), + }) + ``` + +- A local storage adapter (such as `IndexDBFileSystemStorageAdapter` on web) and a remote storage adapter (an implementation of the SDK's `RemoteStorageAdapter`, for example backed by Supabase Storage). Both are generic to all attachment users. See the SDK docs for the available adapters and the remote-adapter contract. + +### 1. Create the attachments collection + +This is the piece that makes the integration TanStack-aware: a normal PowerSync collection over the attachments table. The queue reads and writes attachment records through it. + +```ts +import { createCollection } from "@tanstack/react-db" +import { powerSyncCollectionOptions } from "@tanstack/powersync-db-collection" + +const attachmentsCollection = createCollection( + powerSyncCollectionOptions({ + database: db, + table: APP_SCHEMA.props.attachments, + }) +) +``` + +### 2. Construct the queue + +Pass your collection as `attachmentsCollection` alongside the standard `AttachmentQueue` options. Only `attachmentsCollection` and `watchAttachments` (below) are specific to this package; `db`, `localStorage`, `remoteStorage`, and `errorHandler` are the usual SDK options. + +```ts +import { TanStackDBAttachmentQueue } from "@tanstack/powersync-db-collection" + +const attachmentQueue = new TanStackDBAttachmentQueue({ + db, + attachmentsCollection, // TanStack DB collection over your AttachmentTable + localStorage, // SDK local storage adapter + remoteStorage, // your RemoteStorageAdapter (see SDK docs) + watchAttachments, // see step 3 + errorHandler, // standard AttachmentQueue error handler (see SDK docs) +}) +``` + +Start and stop syncing with the standard `attachmentQueue.startSync()` / `attachmentQueue.stopSync()` lifecycle (see SDK docs), typically inside a React effect or provider. + +### 3. Tell the queue which attachments exist (`watchAttachments`) + +`watchAttachments` reports the set of attachment IDs your data currently references, so the queue knows what to download and what to archive. With TanStack DB you drive it from a live query: emit the initial state, then re-emit the complete set on every change, and clean up on abort. + +```ts +import { + createCollection, + isNull, + liveQueryCollectionOptions, + not, +} from "@tanstack/db" +import { WatchedAttachmentItem } from "@powersync/web" + +const watchAttachments = async (onUpdate, abortSignal) => { + // Every row in your data model that references an attachment. + const livePhotoIds = createCollection( + liveQueryCollectionOptions({ + query: (q) => + q + .from({ document: listsCollection }) + .where(({ document }) => not(isNull(document.photo_id))) + .select(({ document }) => ({ photo_id: document.photo_id })), + }) + ) + + const mapper = (item) => + ({ + id: item.photo_id, + fileExtension: "jpg", + }) satisfies WatchedAttachmentItem + + // 1. Report the initial set of referenced attachment IDs. + const initialState = await livePhotoIds.stateWhenReady() + onUpdate(Array.from(initialState.values()).map(mapper)) + + // 2. Re-emit the whole set on every change (the queue expects the holistic state). + livePhotoIds.subscribeChanges(() => { + onUpdate(livePhotoIds.map(mapper)) + }) + + // 3. Clean up when sync stops. + abortSignal.addEventListener("abort", () => livePhotoIds.cleanup(), { + once: true, + }) +} +``` + +> A `watchAttachmentsFromQuery(...)` convenience helper that collapses this boilerplate into a single call is planned. Until then, use the pattern above. + +### 4. Save an attachment atomically with related data + +`saveFileTanStack` writes the file, inserts the attachment record into your collection, and runs your `updateHook` mutations in the same transaction. Use the hook to insert or update the row that references the new attachment, so both land together or not at all. + +```ts +await attachmentQueue.saveFileTanStack({ + data, // file bytes (ArrayBuffer / base64, per your local adapter) + fileExtension: "jpg", + updateHook: async (attachmentRecord) => { + // Runs in the same transaction as the attachment insert. + listsCollection.insert({ + id: crypto.randomUUID(), + name, + created_at: new Date(), + owner_id: userID, + photo_id: attachmentRecord.id, // associate the row with the attachment + }) + }, +}) +``` + +### 5. Delete an attachment and detach it from the row + +`deleteFileTanStack` queues the file for deletion and runs your `updateHook` in the same transaction. Clear the foreign key so the row and the attachment stay consistent. + +```ts +await attachmentQueue.deleteFileTanStack({ + id: photo_id, + updateHook: async () => { + listsCollection.update(listId, (draft) => { + draft.photo_id = null + }) + }, +}) +``` + +### 6. Display attachments via a live-query join + +Join your attachments collection into a live query to read the local URI (the locally cached file path) alongside your domain rows: + +```ts +import { eq } from "@tanstack/db" + +const { data } = useLiveQuery((q) => + q + .from({ lists: listsCollection }) + .leftJoin({ attachment: attachmentsCollection }, ({ lists, attachment }) => + eq(lists.photo_id, attachment.id) + ) + .select(({ lists, attachment }) => ({ + id: lists.id, + name: lists.name, + photo_id: lists.photo_id, + attachment_local_uri: attachment?.local_uri, + })) +) ``` \ No newline at end of file diff --git a/packages/powersync-db-collection/src/attachments.ts b/packages/powersync-db-collection/src/attachments.ts index c3c9d8f677..cbda937226 100644 --- a/packages/powersync-db-collection/src/attachments.ts +++ b/packages/powersync-db-collection/src/attachments.ts @@ -40,6 +40,10 @@ export interface SaveFileTanStackOptions { export interface DeleteFileTanStackOptions { id: string + /** * + * Note that this is called inside a synchronous TanStackDB transaction, + * any mutations made to other collections will be in the same transaction. + */ updateHook?: (attachment: AttachmentQueueRow) => Promise } From 3770d3b3e87966ec55775cae46e0ea762a136a4c Mon Sep 17 00:00:00 2001 From: Christiaan Landman Date: Tue, 23 Jun 2026 14:01:10 +0200 Subject: [PATCH 04/13] changeset. --- .changeset/curly-planets-lead.md | 6 ++++++ 1 file changed, 6 insertions(+) create mode 100644 .changeset/curly-planets-lead.md diff --git a/.changeset/curly-planets-lead.md b/.changeset/curly-planets-lead.md new file mode 100644 index 0000000000..891ffbb522 --- /dev/null +++ b/.changeset/curly-planets-lead.md @@ -0,0 +1,6 @@ +--- +'@tanstack/powersync-db-collection': patch +--- + +Add attachments support via `TanStackDBAttachmentQueue`. This extends the PowerSync SDK's `AttachmentQueue` and backs it with +a TanStack DB collection, so file uploads/deletes are managed atomically alongside the relational data. From 38af8c59b7766ebb1b9f53941c9916dd503508ea Mon Sep 17 00:00:00 2001 From: Christiaan Landman Date: Wed, 24 Jun 2026 10:45:59 +0200 Subject: [PATCH 05/13] Rename saveFileTanStack => save, deleteFIleTanStack => delete. --- docs/collections/powersync-collection.md | 8 ++++---- .../powersync-db-collection/src/attachments.ts | 13 +++++-------- .../tests/attachments.test.ts | 18 +++++++++--------- 3 files changed, 18 insertions(+), 21 deletions(-) diff --git a/docs/collections/powersync-collection.md b/docs/collections/powersync-collection.md index b484c5bb92..8799f96aa8 100644 --- a/docs/collections/powersync-collection.md +++ b/docs/collections/powersync-collection.md @@ -1212,10 +1212,10 @@ const watchAttachments = async (onUpdate, abortSignal) => { ### 4. Save an attachment atomically with related data -`saveFileTanStack` writes the file, inserts the attachment record into your collection, and runs your `updateHook` mutations in the same transaction. Use the hook to insert or update the row that references the new attachment, so both land together or not at all. +`save` writes the file, inserts the attachment record into your collection, and runs your `updateHook` mutations in the same transaction. Use the hook to insert or update the row that references the new attachment, so both land together or not at all. ```ts -await attachmentQueue.saveFileTanStack({ +await attachmentQueue.save({ data, // file bytes (ArrayBuffer / base64, per your local adapter) fileExtension: "jpg", updateHook: async (attachmentRecord) => { @@ -1233,10 +1233,10 @@ await attachmentQueue.saveFileTanStack({ ### 5. Delete an attachment and detach it from the row -`deleteFileTanStack` queues the file for deletion and runs your `updateHook` in the same transaction. Clear the foreign key so the row and the attachment stay consistent. +`delete` queues the file for deletion and runs your `updateHook` in the same transaction. Clear the foreign key so the row and the attachment stay consistent. ```ts -await attachmentQueue.deleteFileTanStack({ +await attachmentQueue.delete({ id: photo_id, updateHook: async () => { listsCollection.update(listId, (draft) => { diff --git a/packages/powersync-db-collection/src/attachments.ts b/packages/powersync-db-collection/src/attachments.ts index cbda937226..f70f6494ec 100644 --- a/packages/powersync-db-collection/src/attachments.ts +++ b/packages/powersync-db-collection/src/attachments.ts @@ -25,7 +25,7 @@ export type TanStackDBAttachmentQueueOptions = AttachmentQueueOptions & { attachmentsCollection: Collection } -export interface SaveFileTanStackOptions { +export interface SaveOptions { data: AttachmentData fileExtension: string mediaType?: string @@ -38,7 +38,7 @@ export interface SaveFileTanStackOptions { updateHook?: (attachment: AttachmentQueueRow) => Promise } -export interface DeleteFileTanStackOptions { +export interface DeleteOptions { id: string /** * * Note that this is called inside a synchronous TanStackDB transaction, @@ -70,14 +70,14 @@ export class TanStackDBAttachmentQueue extends AttachmentQueue { * Exposes an `updateHook` option which is called inside a TanStackDB transaction, * relational associations with the provided attachment ID should be made in this hook. */ - async saveFileTanStack({ + async save({ data, fileExtension, mediaType, metaData, id, updateHook, - }: SaveFileTanStackOptions): Promise { + }: SaveOptions): Promise { const resolvedId = id ?? (await this.generateAttachmentId()) const filename = `${resolvedId}.${fileExtension}` const localUri = this.localStorage.getLocalUri(filename) @@ -128,10 +128,7 @@ export class TanStackDBAttachmentQueue extends AttachmentQueue { * Exposes an `updateHook` option which is called inside a TanStackDB transaction, * relational associations with the provided attachment ID should be cleaned up in this hook. */ - async deleteFileTanStack({ - id, - updateHook, - }: DeleteFileTanStackOptions): Promise { + async delete({ id, updateHook }: DeleteOptions): Promise { await this.withAttachmentContext(async (ctx) => { const tanStackDBTransaction = createTransaction({ autoCommit: false, diff --git a/packages/powersync-db-collection/tests/attachments.test.ts b/packages/powersync-db-collection/tests/attachments.test.ts index 411c7cd753..e13ea3dbf9 100644 --- a/packages/powersync-db-collection/tests/attachments.test.ts +++ b/packages/powersync-db-collection/tests/attachments.test.ts @@ -210,13 +210,13 @@ describePowerSync(`PowerSync AttachmentQueue (TanStackDB)`, () => { ) } - describe(`saveFileTanStack`, () => { + describe(`save`, () => { it(`writes the local file and inserts a QUEUED_UPLOAD row into the collection`, async () => { const { createQueue, attachmentsCollection, localStorage } = await setup() const queue = createQueue() const data = new Uint8Array(123).fill(42).buffer - const record = await queue.saveFileTanStack({ + const record = await queue.save({ data, fileExtension: `jpg`, mediaType: `image/jpeg`, @@ -245,7 +245,7 @@ describePowerSync(`PowerSync AttachmentQueue (TanStackDB)`, () => { const queue = createQueue() const userId = randomUUID() - const record = await queue.saveFileTanStack({ + const record = await queue.save({ data: createMockJpegBuffer(), fileExtension: `jpg`, updateHook: async (attachment) => { @@ -284,7 +284,7 @@ describePowerSync(`PowerSync AttachmentQueue (TanStackDB)`, () => { await queue.startSync() const userId = randomUUID() - const record = await queue.saveFileTanStack({ + const record = await queue.save({ data: createMockJpegBuffer(), fileExtension: `jpg`, updateHook: async (attachment) => { @@ -313,7 +313,7 @@ describePowerSync(`PowerSync AttachmentQueue (TanStackDB)`, () => { const queue = createQueue() const id = `my-custom-id` - const record = await queue.saveFileTanStack({ + const record = await queue.save({ id, data: createMockJpegBuffer(), fileExtension: `png`, @@ -324,7 +324,7 @@ describePowerSync(`PowerSync AttachmentQueue (TanStackDB)`, () => { }) }) - describe(`deleteFileTanStack`, () => { + describe(`delete file`, () => { it(`queues an existing attachment for deletion and removes the local file`, async () => { const { createQueue, @@ -336,7 +336,7 @@ describePowerSync(`PowerSync AttachmentQueue (TanStackDB)`, () => { await queue.startSync() const userId = randomUUID() - const record = await queue.saveFileTanStack({ + const record = await queue.save({ data: createMockJpegBuffer(), fileExtension: `jpg`, updateHook: async (attachment) => { @@ -355,7 +355,7 @@ describePowerSync(`PowerSync AttachmentQueue (TanStackDB)`, () => { AttachmentState.SYNCED, ) - await queue.deleteFileTanStack({ + await queue.delete({ id: record.id, updateHook: async (attachment) => { usersCollection.update(userId, (draft) => { @@ -389,7 +389,7 @@ describePowerSync(`PowerSync AttachmentQueue (TanStackDB)`, () => { const hook = vi.fn() await expect( - queue.deleteFileTanStack({ id: `does-not-exist`, updateHook: hook }), + queue.delete({ id: `does-not-exist`, updateHook: hook }), ).rejects.toThrow(/not found/i) // The failing transaction must not have run the hook or touched state. From d82ffc2b031e989143f0dde3ad630a0c7ab579a4 Mon Sep 17 00:00:00 2001 From: Christiaan Landman Date: Wed, 24 Jun 2026 16:04:29 +0200 Subject: [PATCH 06/13] Made `updateHook` synchronous and updated `AttachmentQueueRow` typing. --- packages/powersync-db-collection/package.json | 6 ++-- .../src/attachments.ts | 28 +++++++--------- .../tests/attachments.test.ts | 4 +-- pnpm-lock.yaml | 33 +++++++------------ 4 files changed, 29 insertions(+), 42 deletions(-) diff --git a/packages/powersync-db-collection/package.json b/packages/powersync-db-collection/package.json index 9374371824..fde2ae3828 100644 --- a/packages/powersync-db-collection/package.json +++ b/packages/powersync-db-collection/package.json @@ -59,11 +59,11 @@ "p-defer": "^4.0.1" }, "peerDependencies": { - "@powersync/common": "^1.55.0" + "@powersync/common": "^1.57.0" }, "devDependencies": { - "@powersync/common": "1.55.0", - "@powersync/node": "0.18.1", + "@powersync/common": "1.57.0", + "@powersync/node": "0.19.2", "@types/debug": "^4.1.12", "@vitest/coverage-istanbul": "^3.2.4", "better-sqlite3": "^12.6.2" diff --git a/packages/powersync-db-collection/src/attachments.ts b/packages/powersync-db-collection/src/attachments.ts index f70f6494ec..d3c48fb1d8 100644 --- a/packages/powersync-db-collection/src/attachments.ts +++ b/packages/powersync-db-collection/src/attachments.ts @@ -1,8 +1,6 @@ import { AttachmentQueue, - AttachmentState, - AttachmentTable, - Schema, + AttachmentState } from '@powersync/common' import { createTransaction } from '@tanstack/db' import { PowerSyncTransactor } from './PowerSyncTransactor' @@ -11,10 +9,10 @@ import type { AbstractPowerSyncDatabase, AttachmentData, AttachmentQueueOptions, -} from '@powersync/common' -import type { Collection } from '@tanstack/db' -export type AttachmentQueueRow = (typeof _tmpSchema)['types']['attachments'] + AttachmentTable} from '@powersync/common' +import type { Collection } from '@tanstack/db' +import type { OptionalExtractedTable } from './helpers' export type TanStackDBAttachmentQueueOptions = AttachmentQueueOptions & { /** @@ -32,24 +30,22 @@ export interface SaveOptions { metaData?: string id?: string /** - * Note that this is called inside a synchronous TanStackDB transaction, - * any mutations made to other collections will be in the same transaction. + * Called within the same TanStackDB transaction as the attachment write, + * so any mutations made to other collections are committed atomically with it. */ - updateHook?: (attachment: AttachmentQueueRow) => Promise + updateHook?: (attachment: AttachmentQueueRow) => void } export interface DeleteOptions { id: string - /** * - * Note that this is called inside a synchronous TanStackDB transaction, - * any mutations made to other collections will be in the same transaction. + /** + * Called within the same TanStackDB transaction as the attachment write, + * so any mutations made to other collections are committed atomically with it. */ - updateHook?: (attachment: AttachmentQueueRow) => Promise + updateHook?: (attachment: AttachmentQueueRow) => void } -const _tmpSchema = new Schema({ - attachments: new AttachmentTable(), -}) +export type AttachmentQueueRow = OptionalExtractedTable /** * A custom extension of the PowerSyncAttachmentQueue for TanStackDB. diff --git a/packages/powersync-db-collection/tests/attachments.test.ts b/packages/powersync-db-collection/tests/attachments.test.ts index e13ea3dbf9..c5925e234f 100644 --- a/packages/powersync-db-collection/tests/attachments.test.ts +++ b/packages/powersync-db-collection/tests/attachments.test.ts @@ -229,7 +229,7 @@ describePowerSync(`PowerSync AttachmentQueue (TanStackDB)`, () => { expect(record.has_synced).toBe(0) // The file should exist on disk at the returned local_uri. - expect(await localStorage.fileExists(record.local_uri)).toBe(true) + expect(await localStorage.fileExists(record.local_uri!)).toBe(true) // The row should be reflected in the collection once it syncs back. await waitForState( @@ -379,7 +379,7 @@ describePowerSync(`PowerSync AttachmentQueue (TanStackDB)`, () => { () => expect(attachmentsCollection.get(record.id)).toBeUndefined(), { timeout: WAIT_TIMEOUT, interval: 50 }, ) - expect(await localStorage.fileExists(record.local_uri)).toBe(false) + expect(await localStorage.fileExists(record.local_uri!)).toBe(false) }) it(`throws for an unknown id and commits nothing`, async () => { diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 3852ece3ed..6e61ea7bdd 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1338,11 +1338,11 @@ importers: version: 4.0.1 devDependencies: '@powersync/common': - specifier: 1.55.0 - version: 1.55.0 + specifier: 1.57.0 + version: 1.57.0 '@powersync/node': - specifier: 0.18.1 - version: 0.18.1(@powersync/common@1.55.0)(better-sqlite3@12.8.0) + specifier: 0.19.2 + version: 0.19.2(@powersync/common@1.57.0)(better-sqlite3@12.8.0) '@types/debug': specifier: ^4.1.12 version: 4.1.12 @@ -4831,13 +4831,13 @@ packages: '@poppinss/exception@1.2.3': resolution: {integrity: sha512-dCED+QRChTVatE9ibtoaxc+WkdzOSjYTKi/+uacHWIsfodVfpsueo3+DKpgU5Px8qXjgmXkSvhXvSCz3fnP9lw==} - '@powersync/common@1.55.0': - resolution: {integrity: sha512-c9K2Gac9wOB4ijVnQT388g7Yeuh6VrfJZFXaL6Rag9Fp7F8JzAYcrWaULLTdMnzm7VW6b6XT5M2ip5yLS8oSBg==} + '@powersync/common@1.57.0': + resolution: {integrity: sha512-uYccCxK5mwahELRouY3YY584TZgjFU8wPPKZQQ6sAOUoMikV8D/+v+UYsNI280MKMnhFqLkxk4TPZIG7ArIzTQ==} - '@powersync/node@0.18.1': - resolution: {integrity: sha512-fcTICgs61CAEb39xiC7pedYsPgbjUInJ/47dr7RIdnEHpAgjWH8bW95/b70qK1fQUANy9lKBBF3PcmfswVgfCw==} + '@powersync/node@0.19.2': + resolution: {integrity: sha512-lF7v/rkiLujAojn7Vjgvs1AibhL5zlEQVYO0iCUGoE1S1Hw7lxfUvAa1mTneKWCEmj0EC9yQBHkPUyBDZXVdLA==} peerDependencies: - '@powersync/common': ^1.49.0 + '@powersync/common': ^1.57.0 better-sqlite3: 12.x peerDependenciesMeta: better-sqlite3: @@ -6799,9 +6799,6 @@ packages: async-limiter@1.0.1: resolution: {integrity: sha512-csOlWGAcRFJaI6m+F2WKdnMKr4HhdhFVBk0H/QbJFMCr+uO2kwohwXQPxw/9OCxp05r5ghVBFSyioixx3gfkNQ==} - async-mutex@0.5.0: - resolution: {integrity: sha512-1A94B18jkJ3DYq284ohPxoXbfTA5HsQ7/Mf4DEhcyLx3Bz27Rh59iScbB6EPiP+B+joue6YCxcMXSbFC1tZKwA==} - asynckit@0.4.0: resolution: {integrity: sha512-Oei9OH4tRh0YqU3GxhX79dM/mwVgvbZJaSNaRk+bshkj0S5cfHcgYakreBjrHwatXKbz+IoIdYLxrKim2MjW0Q==} @@ -16480,15 +16477,13 @@ snapshots: '@poppinss/exception@1.2.3': {} - '@powersync/common@1.55.0': + '@powersync/common@1.57.0': dependencies: js-logger: 1.6.1 - '@powersync/node@0.18.1(@powersync/common@1.55.0)(better-sqlite3@12.8.0)': + '@powersync/node@0.19.2(@powersync/common@1.57.0)(better-sqlite3@12.8.0)': dependencies: - '@powersync/common': 1.55.0 - async-mutex: 0.5.0 - bson: 6.10.4 + '@powersync/common': 1.57.0 comlink: 4.4.2 undici: 7.24.4 optionalDependencies: @@ -18956,10 +18951,6 @@ snapshots: async-limiter@1.0.1: {} - async-mutex@0.5.0: - dependencies: - tslib: 2.8.1 - asynckit@0.4.0: {} at-least-node@1.0.0: {} From dc7f13e7430ce194ad6a6d5a909bcdc970e7b173 Mon Sep 17 00:00:00 2001 From: Christiaan Landman Date: Tue, 28 Jul 2026 15:27:16 +0200 Subject: [PATCH 07/13] Minor patch changeset. Updated docs to reflec sync nature of updateHook. --- .changeset/curly-planets-lead.md | 2 +- docs/collections/powersync-collection.md | 8 +++++--- 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/.changeset/curly-planets-lead.md b/.changeset/curly-planets-lead.md index 891ffbb522..890d4c36b5 100644 --- a/.changeset/curly-planets-lead.md +++ b/.changeset/curly-planets-lead.md @@ -1,5 +1,5 @@ --- -'@tanstack/powersync-db-collection': patch +'@tanstack/powersync-db-collection': minor --- Add attachments support via `TanStackDBAttachmentQueue`. This extends the PowerSync SDK's `AttachmentQueue` and backs it with diff --git a/docs/collections/powersync-collection.md b/docs/collections/powersync-collection.md index 8799f96aa8..983d38e829 100644 --- a/docs/collections/powersync-collection.md +++ b/docs/collections/powersync-collection.md @@ -1218,7 +1218,7 @@ const watchAttachments = async (onUpdate, abortSignal) => { await attachmentQueue.save({ data, // file bytes (ArrayBuffer / base64, per your local adapter) fileExtension: "jpg", - updateHook: async (attachmentRecord) => { + updateHook: (attachmentRecord) => { // Runs in the same transaction as the attachment insert. listsCollection.insert({ id: crypto.randomUUID(), @@ -1231,14 +1231,16 @@ await attachmentQueue.save({ }) ``` +> `updateHook` must be synchronous, it runs inside the transaction's synchronous `mutate()` block and its return value is not awaited, so any mutation after an `await` escapes the transaction. Do asynchronous work before calling `save` or `delete`. + ### 5. Delete an attachment and detach it from the row -`delete` queues the file for deletion and runs your `updateHook` in the same transaction. Clear the foreign key so the row and the attachment stay consistent. +`delete` queues the file for deletion and runs your `updateHook` in the same transaction. Clear the foreign key so the row and the attachment stay consistent. As with `save`, the hook must be synchronous. ```ts await attachmentQueue.delete({ id: photo_id, - updateHook: async () => { + updateHook: () => { listsCollection.update(listId, (draft) => { draft.photo_id = null }) From 6d458a8b2147f974527810ee96094ebe5e4f4e97 Mon Sep 17 00:00:00 2001 From: Christiaan Landman Date: Mon, 3 Aug 2026 10:26:01 +0200 Subject: [PATCH 08/13] Rollback new written attachment on transaction failure. --- .../src/attachments.ts | 104 +++++++++++++----- .../tests/attachments.test.ts | 98 ++++++++++++++++- 2 files changed, 173 insertions(+), 29 deletions(-) diff --git a/packages/powersync-db-collection/src/attachments.ts b/packages/powersync-db-collection/src/attachments.ts index d3c48fb1d8..35117fbd72 100644 --- a/packages/powersync-db-collection/src/attachments.ts +++ b/packages/powersync-db-collection/src/attachments.ts @@ -11,7 +11,7 @@ import type { AttachmentQueueOptions, AttachmentTable} from '@powersync/common' -import type { Collection } from '@tanstack/db' +import type { Collection, Transaction } from '@tanstack/db' import type { OptionalExtractedTable } from './helpers' export type TanStackDBAttachmentQueueOptions = AttachmentQueueOptions & { @@ -28,6 +28,13 @@ export interface SaveOptions { fileExtension: string mediaType?: string metaData?: string + /** + * Optional custom ID. If not provided, a UUID will be generated. + * + * Reusing the ID of an existing attachment overwrites that attachment's local file + * before the write is rejected, and the file is then removed by cleanup — leaving the + * existing record without its file. Pass an ID that is not already in the queue. + */ id?: string /** * Called within the same TanStackDB transaction as the attachment write, @@ -91,29 +98,36 @@ export class TanStackDBAttachmentQueue extends AttachmentQueue { meta_data: metaData ?? null, } - /** - * We use the attachmentService lock to prevent attachment queue race conditions — specifically, - * it stops the watcher from treating a newly inserted attachment record as one that needs - * to be downloaded. - * */ - await this.withAttachmentContext(async (ctx) => { - const tanStackDBTransaction = createTransaction({ - autoCommit: false, - mutationFn: async ({ transaction }) => { - await new PowerSyncTransactor({ - database: ctx.db, - }).applyTransaction(transaction) - }, - }) + try { + /** + * We use the attachmentService lock to prevent attachment queue race conditions — specifically, + * it stops the watcher from treating a newly inserted attachment record as one that needs + * to be downloaded. + * */ + await this.withAttachmentContext(async (ctx) => { + const tanStackDBTransaction = createTransaction({ + autoCommit: false, + mutationFn: async ({ transaction }) => { + await new PowerSyncTransactor({ + database: ctx.db, + }).applyTransaction(transaction) + }, + }) - tanStackDBTransaction.mutate(() => { - this.collection.insert(attachment) - // allow the user to associate values in this transaction - updateHook?.(attachment) + await this.runInTransaction(tanStackDBTransaction, () => { + this.collection.insert(attachment) + // allow the user to associate values in this transaction + updateHook?.(attachment) + }) }) - - await tanStackDBTransaction.commit() - }) + } catch (error) { + /** + * The file is written before the transaction opens, so a failed transaction would + * otherwise leave an orphaned file behind that no attachment record points to. + */ + await this.deleteLocalFile(localUri) + throw error + } return attachment } @@ -135,7 +149,7 @@ export class TanStackDBAttachmentQueue extends AttachmentQueue { }, }) - tanStackDBTransaction.mutate(() => { + await this.runInTransaction(tanStackDBTransaction, () => { const attachment = this.collection.get(id) if (!attachment) { throw new Error(`Attachment with id ${id} not found`) @@ -149,8 +163,48 @@ export class TanStackDBAttachmentQueue extends AttachmentQueue { // allow the user to associate values in this transaction updateHook?.(attachment) }) - - await tanStackDBTransaction.commit() }) } + + /** + * Applies `mutations` to `transaction` and commits it, rolling back on any failure. + * + * `Transaction.mutate` does not roll back when its callback throws, so a throwing + * `updateHook` would otherwise leave the transaction pending with its optimistic + * mutations still applied to the collections. + */ + protected async runInTransaction( + transaction: Transaction, + mutations: () => void, + ): Promise { + /** + * `rollback` rejects this promise. The error is already surfaced to the caller by the + * throw below, so this catch only stops it from becoming an unhandled rejection. + */ + void transaction.isPersisted.promise.catch(() => {}) + + try { + transaction.mutate(mutations) + } catch (error) { + transaction.rollback() + throw error + } + + await transaction.commit() + } + + /** + * Best-effort removal of a local file. A cleanup failure is logged rather than thrown, + * so that it can never mask the error which triggered the cleanup. + */ + protected async deleteLocalFile(localUri: string): Promise { + try { + await this.localStorage.deleteFile(localUri) + } catch (error) { + this.logger.error( + `Could not clean up local attachment file ${localUri}`, + error, + ) + } + } } diff --git a/packages/powersync-db-collection/tests/attachments.test.ts b/packages/powersync-db-collection/tests/attachments.test.ts index c5925e234f..c7af2f8b2e 100644 --- a/packages/powersync-db-collection/tests/attachments.test.ts +++ b/packages/powersync-db-collection/tests/attachments.test.ts @@ -248,7 +248,7 @@ describePowerSync(`PowerSync AttachmentQueue (TanStackDB)`, () => { const record = await queue.save({ data: createMockJpegBuffer(), fileExtension: `jpg`, - updateHook: async (attachment) => { + updateHook: (attachment) => { usersCollection.insert({ id: userId, name: `steven`, @@ -287,7 +287,7 @@ describePowerSync(`PowerSync AttachmentQueue (TanStackDB)`, () => { const record = await queue.save({ data: createMockJpegBuffer(), fileExtension: `jpg`, - updateHook: async (attachment) => { + updateHook: (attachment) => { usersCollection.insert({ id: userId, name: `steven`, @@ -322,6 +322,44 @@ describePowerSync(`PowerSync AttachmentQueue (TanStackDB)`, () => { expect(record.id).toBe(id) expect(record.filename).toBe(`${id}.png`) }) + + it(`removes the local file and rolls back when the updateHook throws`, async () => { + const { + createQueue, + attachmentsCollection, + usersCollection, + localStorage, + } = await setup() + const queue = createQueue() + + // A caller-supplied id lets us derive the local uri without a returned record. + const id = randomUUID() + const localUri = localStorage.getLocalUri(`${id}.jpg`) + + await expect( + queue.save({ + id, + data: createMockJpegBuffer(), + fileExtension: `jpg`, + updateHook: (attachment) => { + usersCollection.insert({ + id: randomUUID(), + name: `steven`, + email: `steven@journeyapps.com`, + photo_id: attachment.id, + }) + throw new Error(`updateHook failed`) + }, + }), + ).rejects.toThrow(/updateHook failed/) + + // The file is written before the transaction opens, so it must be cleaned up. + expect(await localStorage.fileExists(localUri)).toBe(false) + + // Neither the attachment nor the hook's own mutation may survive the failure. + expect(attachmentsCollection.get(id)).toBeUndefined() + expect(usersCollection.size).toBe(0) + }) }) describe(`delete file`, () => { @@ -339,7 +377,7 @@ describePowerSync(`PowerSync AttachmentQueue (TanStackDB)`, () => { const record = await queue.save({ data: createMockJpegBuffer(), fileExtension: `jpg`, - updateHook: async (attachment) => { + updateHook: (attachment) => { usersCollection.insert({ id: userId, name: `steven`, @@ -357,7 +395,7 @@ describePowerSync(`PowerSync AttachmentQueue (TanStackDB)`, () => { await queue.delete({ id: record.id, - updateHook: async (attachment) => { + updateHook: (attachment) => { usersCollection.update(userId, (draft) => { if (draft.photo_id === attachment.id) { draft.photo_id = null @@ -397,5 +435,57 @@ describePowerSync(`PowerSync AttachmentQueue (TanStackDB)`, () => { expect(attachmentsCollection.get(`does-not-exist`)).toBeUndefined() expect(usersCollection.size).toBe(0) }) + + it(`rolls back the queued deletion when the updateHook throws`, async () => { + const { + createQueue, + attachmentsCollection, + usersCollection, + localStorage, + } = await setup() + const queue = createQueue() + + const userId = randomUUID() + const record = await queue.save({ + data: createMockJpegBuffer(), + fileExtension: `jpg`, + updateHook: (attachment) => { + usersCollection.insert({ + id: userId, + name: `steven`, + email: `steven@journeyapps.com`, + photo_id: attachment.id, + }) + }, + }) + + // Sync is deliberately left stopped: the rollback happens entirely in the + // foreground transaction, and a running sync loop would only race teardown. + await waitForState( + attachmentsCollection, + record.id, + AttachmentState.QUEUED_UPLOAD, + ) + + await expect( + queue.delete({ + id: record.id, + updateHook: () => { + usersCollection.update(userId, (draft) => { + draft.photo_id = null + }) + throw new Error(`updateHook failed`) + }, + }), + ).rejects.toThrow(/updateHook failed/) + + // Both the QUEUED_DELETE transition and the hook's mutation must be rolled back, + // leaving the attachment intact rather than half-deleted. + expect(attachmentsCollection.get(record.id)?.state).toBe( + AttachmentState.QUEUED_UPLOAD, + ) + expect(usersCollection.get(userId)?.photo_id).toBe(record.id) + expect(await localStorage.fileExists(record.local_uri!)).toBe(true) + }) }) }) From f9f68230b7494cb3eb7e6199fa93b5e7b49a2b90 Mon Sep 17 00:00:00 2001 From: Christiaan Landman Date: Mon, 3 Aug 2026 11:14:20 +0200 Subject: [PATCH 09/13] Cleanup import formatting and removed incorrect doc line. --- docs/collections/powersync-collection.md | 2 -- packages/powersync-db-collection/src/attachments.ts | 9 +++------ 2 files changed, 3 insertions(+), 8 deletions(-) diff --git a/docs/collections/powersync-collection.md b/docs/collections/powersync-collection.md index 983d38e829..7982a8f376 100644 --- a/docs/collections/powersync-collection.md +++ b/docs/collections/powersync-collection.md @@ -1208,8 +1208,6 @@ const watchAttachments = async (onUpdate, abortSignal) => { } ``` -> A `watchAttachmentsFromQuery(...)` convenience helper that collapses this boilerplate into a single call is planned. Until then, use the pattern above. - ### 4. Save an attachment atomically with related data `save` writes the file, inserts the attachment record into your collection, and runs your `updateHook` mutations in the same transaction. Use the hook to insert or update the row that references the new attachment, so both land together or not at all. diff --git a/packages/powersync-db-collection/src/attachments.ts b/packages/powersync-db-collection/src/attachments.ts index 35117fbd72..6744e2aaae 100644 --- a/packages/powersync-db-collection/src/attachments.ts +++ b/packages/powersync-db-collection/src/attachments.ts @@ -1,7 +1,4 @@ -import { - AttachmentQueue, - AttachmentState -} from '@powersync/common' +import { AttachmentQueue, AttachmentState } from '@powersync/common' import { createTransaction } from '@tanstack/db' import { PowerSyncTransactor } from './PowerSyncTransactor' @@ -9,8 +6,8 @@ import type { AbstractPowerSyncDatabase, AttachmentData, AttachmentQueueOptions, - - AttachmentTable} from '@powersync/common' + AttachmentTable, +} from '@powersync/common' import type { Collection, Transaction } from '@tanstack/db' import type { OptionalExtractedTable } from './helpers' From 5d3b5000fd831d953b9088a9811e492301a98ddb Mon Sep 17 00:00:00 2001 From: Christiaan Landman Date: Mon, 3 Aug 2026 11:39:28 +0200 Subject: [PATCH 10/13] Minor coderabbit feedback. --- .../src/attachments.ts | 28 +++++++++++++++---- .../tests/attachments.test.ts | 24 ++++++++++++++++ 2 files changed, 47 insertions(+), 5 deletions(-) diff --git a/packages/powersync-db-collection/src/attachments.ts b/packages/powersync-db-collection/src/attachments.ts index 6744e2aaae..b17daabe80 100644 --- a/packages/powersync-db-collection/src/attachments.ts +++ b/packages/powersync-db-collection/src/attachments.ts @@ -28,14 +28,17 @@ export interface SaveOptions { /** * Optional custom ID. If not provided, a UUID will be generated. * - * Reusing the ID of an existing attachment overwrites that attachment's local file - * before the write is rejected, and the file is then removed by cleanup — leaving the - * existing record without its file. Pass an ID that is not already in the queue. + * Rejected if an attachment with this ID is already in the queue. */ id?: string /** - * Called within the same TanStackDB transaction as the attachment write, + * Called synchronously within the same TanStackDB transaction as the attachment write, * so any mutations made to other collections are committed atomically with it. + * + * Must not be async. `Transaction.mutate` unregisters the ambient transaction as soon as + * this callback returns, so any mutation made after an `await` inside the hook escapes the + * transaction and is not committed atomically with the attachment. Do asynchronous work + * before calling `save` or `delete`. */ updateHook?: (attachment: AttachmentQueueRow) => void } @@ -43,8 +46,13 @@ export interface SaveOptions { export interface DeleteOptions { id: string /** - * Called within the same TanStackDB transaction as the attachment write, + * Called synchronously within the same TanStackDB transaction as the attachment write, * so any mutations made to other collections are committed atomically with it. + * + * Must not be async. `Transaction.mutate` unregisters the ambient transaction as soon as + * this callback returns, so any mutation made after an `await` inside the hook escapes the + * transaction and is not committed atomically with the attachment. Do asynchronous work + * before calling `save` or `delete`. */ updateHook?: (attachment: AttachmentQueueRow) => void } @@ -79,6 +87,16 @@ export class TanStackDBAttachmentQueue extends AttachmentQueue { updateHook, }: SaveOptions): Promise { const resolvedId = id ?? (await this.generateAttachmentId()) + + /** + * Checked before the file is written. Writing first would overwrite the existing + * attachment's local file, and the cleanup below would then delete it — leaving the + * pre-existing record pointing at a file that no longer exists. + */ + if (this.collection.get(resolvedId)) { + throw new Error(`Attachment with id ${resolvedId} already exists`) + } + const filename = `${resolvedId}.${fileExtension}` const localUri = this.localStorage.getLocalUri(filename) const size = await this.localStorage.saveFile(localUri, data) diff --git a/packages/powersync-db-collection/tests/attachments.test.ts b/packages/powersync-db-collection/tests/attachments.test.ts index c7af2f8b2e..218b93d2b8 100644 --- a/packages/powersync-db-collection/tests/attachments.test.ts +++ b/packages/powersync-db-collection/tests/attachments.test.ts @@ -323,6 +323,30 @@ describePowerSync(`PowerSync AttachmentQueue (TanStackDB)`, () => { expect(record.filename).toBe(`${id}.png`) }) + it(`rejects a reused id without disturbing the existing attachment`, async () => { + const { createQueue, attachmentsCollection, localStorage } = await setup() + const queue = createQueue() + + const original = await queue.save({ + data: createMockJpegBuffer(), + fileExtension: `jpg`, + }) + + await expect( + queue.save({ + id: original.id, + // A different payload, so an overwrite would be detectable by size alone. + data: new Uint8Array(999).fill(7).buffer, + fileExtension: `jpg`, + }), + ).rejects.toThrow(/already exists/) + + // Without the up-front check the reused id overwrites this file and cleanup + // then deletes it, leaving the original record pointing at nothing. + expect(await localStorage.fileExists(original.local_uri!)).toBe(true) + expect(attachmentsCollection.get(original.id)?.size).toBe(original.size) + }) + it(`removes the local file and rolls back when the updateHook throws`, async () => { const { createQueue, From fc55bee66c05376e594aa7e377ab0a96f7087405 Mon Sep 17 00:00:00 2001 From: Christiaan Landman Date: Tue, 11 Aug 2026 15:06:54 +0200 Subject: [PATCH 11/13] Address concurrent save race issue. --- .../src/attachments.ts | 78 +++++++++---------- .../tests/attachments.test.ts | 42 ++++++++++ 2 files changed, 80 insertions(+), 40 deletions(-) diff --git a/packages/powersync-db-collection/src/attachments.ts b/packages/powersync-db-collection/src/attachments.ts index b17daabe80..7fb7dd926b 100644 --- a/packages/powersync-db-collection/src/attachments.ts +++ b/packages/powersync-db-collection/src/attachments.ts @@ -87,39 +87,37 @@ export class TanStackDBAttachmentQueue extends AttachmentQueue { updateHook, }: SaveOptions): Promise { const resolvedId = id ?? (await this.generateAttachmentId()) - - /** - * Checked before the file is written. Writing first would overwrite the existing - * attachment's local file, and the cleanup below would then delete it — leaving the - * pre-existing record pointing at a file that no longer exists. - */ - if (this.collection.get(resolvedId)) { - throw new Error(`Attachment with id ${resolvedId} already exists`) - } - const filename = `${resolvedId}.${fileExtension}` const localUri = this.localStorage.getLocalUri(filename) - const size = await this.localStorage.saveFile(localUri, data) - - const attachment: AttachmentQueueRow = { - id: resolvedId, - filename, - media_type: mediaType ?? null, - local_uri: localUri, - state: AttachmentState.QUEUED_UPLOAD, - has_synced: 0, - size, - timestamp: new Date().getTime(), - meta_data: metaData ?? null, - } - try { + return this.withAttachmentContext(async (ctx) => { /** - * We use the attachmentService lock to prevent attachment queue race conditions — specifically, - * it stops the watcher from treating a newly inserted attachment record as one that needs - * to be downloaded. - * */ - await this.withAttachmentContext(async (ctx) => { + * Checked before the file is written. Writing first would overwrite the existing + * attachment's local file, and the cleanup below would then delete it — leaving the + * pre-existing record pointing at a file that no longer exists. + * + * Deliberately outside the `try`: this throw must not reach the cleanup, because the file + * at `localUri` belongs to the pre-existing attachment rather than to this call. + */ + if (this.collection.get(resolvedId)) { + throw new Error(`Attachment with id ${resolvedId} already exists`) + } + + const size = await this.localStorage.saveFile(localUri, data) + + const attachment: AttachmentQueueRow = { + id: resolvedId, + filename, + media_type: mediaType ?? null, + local_uri: localUri, + state: AttachmentState.QUEUED_UPLOAD, + has_synced: 0, + size, + timestamp: new Date().getTime(), + meta_data: metaData ?? null, + } + + try { const tanStackDBTransaction = createTransaction({ autoCommit: false, mutationFn: async ({ transaction }) => { @@ -134,17 +132,17 @@ export class TanStackDBAttachmentQueue extends AttachmentQueue { // allow the user to associate values in this transaction updateHook?.(attachment) }) - }) - } catch (error) { - /** - * The file is written before the transaction opens, so a failed transaction would - * otherwise leave an orphaned file behind that no attachment record points to. - */ - await this.deleteLocalFile(localUri) - throw error - } - - return attachment + } catch (error) { + /** + * The file is written before the transaction opens, so a failed transaction would + * otherwise leave an orphaned file behind that no attachment record points to. + */ + await this.deleteLocalFile(localUri) + throw error + } + + return attachment + }) } /** diff --git a/packages/powersync-db-collection/tests/attachments.test.ts b/packages/powersync-db-collection/tests/attachments.test.ts index 218b93d2b8..193f8fa0c1 100644 --- a/packages/powersync-db-collection/tests/attachments.test.ts +++ b/packages/powersync-db-collection/tests/attachments.test.ts @@ -24,6 +24,7 @@ import type { RemoteStorageAdapter, WatchedAttachmentItem, } from '@powersync/common' +import type { AttachmentQueueRow } from '../src/attachments' // A minimal valid 1x1 pixel JPEG used as the remote payload for downloads. const MOCK_JPEG_U8A = [ @@ -347,6 +348,47 @@ describePowerSync(`PowerSync AttachmentQueue (TanStackDB)`, () => { expect(attachmentsCollection.get(original.id)?.size).toBe(original.size) }) + it(`keeps the winner's file intact when two saves race on the same id`, async () => { + const { createQueue, attachmentsCollection, localStorage } = await setup() + const queue = createQueue() + + const id = `contended-id` + // Distinct payload sizes, so an overwrite is detectable from the file length + const smallPayload = createMockJpegBuffer() + const largePayload = new Uint8Array(999).fill(7).buffer + expect(smallPayload.byteLength).not.toBe(largePayload.byteLength) + + // Both calls run their duplicate check before either has inserted + const results = await Promise.allSettled([ + queue.save({ id, data: smallPayload, fileExtension: `jpg` }), + queue.save({ id, data: largePayload, fileExtension: `jpg` }), + ]) + + const fulfilled = results.filter( + (result): result is PromiseFulfilledResult => + result.status === `fulfilled`, + ) + const rejected = results.filter( + (result): result is PromiseRejectedResult => + result.status === `rejected`, + ) + + expect(fulfilled).toHaveLength(1) + expect(rejected).toHaveLength(1) + expect(rejected[0]!.reason).toEqual( + expect.objectContaining({ message: expect.stringMatching(/exists/) }), + ) + + const winner = fulfilled[0]!.value + expect(attachmentsCollection.size).toBe(1) + expect(attachmentsCollection.get(id)?.size).toBe(winner.size) + + // The loser's cleanup must not delete the file the winner's record points at + expect(await localStorage.fileExists(winner.local_uri!)).toBe(true) + const onDisk = await localStorage.readFile(winner.local_uri!) + expect(onDisk.byteLength).toBe(winner.size) + }) + it(`removes the local file and rolls back when the updateHook throws`, async () => { const { createQueue, From bb7cbbae7379dcaf047ef85e68731a8633b4d0be Mon Sep 17 00:00:00 2001 From: Kyle Mathews Date: Thu, 10 Sep 2026 21:39:55 -0600 Subject: [PATCH 12/13] fix(powersync): load attachment IDs before mutations and guard file ownership Hold a targeted live-query demand through transaction confirmation for eager and on-demand collections. Reject same-ID overlapping saves across queues sharing a database and preserve SDK filename-based relocation. Add intent oracles and separate native/integration repros for the unresolved upstream upload-completion race. --- .changeset/curly-planets-lead.md | 2 +- docs/collections/powersync-collection.md | 10 +- packages/powersync-db-collection/package.json | 3 +- .../src/attachments.ts | 188 ++++--- .../tests/ATTACHMENT-ORACLE.md | 37 ++ .../tests/attachments-lifecycle-fixture.ts | 467 ++++++++++++++++++ .../attachments-lifecycle-oracle.test.ts | 37 ++ .../tests/attachments-native-sdk.repro.ts | 85 ++++ .../tests/attachments-sdk-completion.repro.ts | 30 ++ .../tests/attachments.test.ts | 405 ++++++++++++++- .../tests/upstream.config.ts | 12 + 11 files changed, 1193 insertions(+), 83 deletions(-) create mode 100644 packages/powersync-db-collection/tests/ATTACHMENT-ORACLE.md create mode 100644 packages/powersync-db-collection/tests/attachments-lifecycle-fixture.ts create mode 100644 packages/powersync-db-collection/tests/attachments-lifecycle-oracle.test.ts create mode 100644 packages/powersync-db-collection/tests/attachments-native-sdk.repro.ts create mode 100644 packages/powersync-db-collection/tests/attachments-sdk-completion.repro.ts create mode 100644 packages/powersync-db-collection/tests/upstream.config.ts diff --git a/.changeset/curly-planets-lead.md b/.changeset/curly-planets-lead.md index 890d4c36b5..8f0b2cbeb7 100644 --- a/.changeset/curly-planets-lead.md +++ b/.changeset/curly-planets-lead.md @@ -3,4 +3,4 @@ --- Add attachments support via `TanStackDBAttachmentQueue`. This extends the PowerSync SDK's `AttachmentQueue` and backs it with -a TanStack DB collection, so file uploads/deletes are managed atomically alongside the relational data. +a TanStack DB collection, so attachment metadata and related rows commit atomically. Local files and remote uploads/deletes are managed separately. diff --git a/docs/collections/powersync-collection.md b/docs/collections/powersync-collection.md index 7982a8f376..0ffaa382be 100644 --- a/docs/collections/powersync-collection.md +++ b/docs/collections/powersync-collection.md @@ -1103,7 +1103,7 @@ const liveQuery = createLiveQueryCollection({ ## Attachments -`@tanstack/powersync-db-collection` ships `TanStackDBAttachmentQueue`, an [`AttachmentQueue`](https://docs.powersync.com/usage/use-case-examples/attachments-files) whose file operations commit inside a TanStack DB collection transaction. This lets you create (or delete) an attachment and mutate a related collection row (for example, setting `lists.photo_id`) atomically in a single transaction, instead of issuing two independent writes. +`@tanstack/powersync-db-collection` ships `TanStackDBAttachmentQueue`, an [`AttachmentQueue`](https://docs.powersync.com/usage/use-case-examples/attachments-files) that commits attachment metadata and related collection mutations (for example, setting `lists.photo_id`) in one database transaction. File I/O is separate: a failed save attempts to remove its local file, while the SDK performs remote uploads and deletes later. The queue extends PowerSync's `AttachmentQueue`, so the generic concepts are unchanged and documented once in the SDK. @@ -1130,6 +1130,10 @@ These are standard PowerSync attachment requirements. See the SDK attachments do This is the piece that makes the integration TanStack-aware: a normal PowerSync collection over the attachments table. The queue reads and writes attachment records through it. +Both eager and on-demand collections work. Before `save` or `delete` opens its mutation, the queue loads the attachment ID through a temporary live query and retains that query until the transaction is confirmed. It does not call `preload()` inside a mutation function or require loading the entire table. + +An existing ID, or a concurrent save of that ID through the same PowerSync database object, is rejected before writing the file. This is an in-process guard, not a lock across separate database handles, SDK queues, tabs, or processes. File names retain the SDK's ID-based convention so restart can find files after the app's storage directory moves. + ```ts import { createCollection } from "@tanstack/react-db" import { powerSyncCollectionOptions } from "@tanstack/powersync-db-collection" @@ -1235,6 +1239,8 @@ await attachmentQueue.save({ `delete` queues the file for deletion and runs your `updateHook` in the same transaction. Clear the foreign key so the row and the attachment stay consistent. As with `save`, the hook must be synchronous. +**Upstream limitation:** the SDK version used by this PR can overwrite a queued deletion when an already-running upload succeeds or fails. The related row is detached, but the SDK can lose the remote deletion or retry the obsolete upload. This integration does not work around that SDK completion race. The [attachment oracle notes](../../packages/powersync-db-collection/tests/ATTACHMENT-ORACLE.md) include runnable native-SDK and integration repros; a green ordinary suite does not establish safety for this overlap. + ```ts await attachmentQueue.delete({ id: photo_id, @@ -1266,4 +1272,4 @@ const { data } = useLiveQuery((q) => attachment_local_uri: attachment?.local_uri, })) ) -``` \ No newline at end of file +``` diff --git a/packages/powersync-db-collection/package.json b/packages/powersync-db-collection/package.json index 5f41c617f7..476fb3a50a 100644 --- a/packages/powersync-db-collection/package.json +++ b/packages/powersync-db-collection/package.json @@ -27,7 +27,8 @@ "build": "vite build", "dev": "vite build --watch", "lint": "eslint . --fix", - "test": "vitest --run" + "test": "vitest --run", + "test:upstream-repros": "vitest run --config tests/upstream.config.ts --maxWorkers=2" }, "type": "module", "main": "dist/cjs/index.cjs", diff --git a/packages/powersync-db-collection/src/attachments.ts b/packages/powersync-db-collection/src/attachments.ts index 7fb7dd926b..465d45d13a 100644 --- a/packages/powersync-db-collection/src/attachments.ts +++ b/packages/powersync-db-collection/src/attachments.ts @@ -1,5 +1,9 @@ -import { AttachmentQueue, AttachmentState } from '@powersync/common' -import { createTransaction } from '@tanstack/db' +import { + AttachmentQueue, + AttachmentState, + sanitizeSQL, +} from '@powersync/common' +import { createLiveQueryCollection, createTransaction, eq } from '@tanstack/db' import { PowerSyncTransactor } from './PowerSyncTransactor' import type { @@ -11,6 +15,10 @@ import type { import type { Collection, Transaction } from '@tanstack/db' import type { OptionalExtractedTable } from './helpers' +// SDK context locks belong to individual queues. Reserve saves across queues +// sharing this database so a rejected insert cannot remove another save's file. +const savingIds = new WeakMap>() + export type TanStackDBAttachmentQueueOptions = AttachmentQueueOptions & { /** * For TanStack, we want access to the synced TanStackDB collection. @@ -28,7 +36,8 @@ export interface SaveOptions { /** * Optional custom ID. If not provided, a UUID will be generated. * - * Rejected if an attachment with this ID is already in the queue. + * Rejected if this ID is already in the queue or is being saved by another + * call sharing the same PowerSync database object. */ id?: string /** @@ -89,35 +98,80 @@ export class TanStackDBAttachmentQueue extends AttachmentQueue { const resolvedId = id ?? (await this.generateAttachmentId()) const filename = `${resolvedId}.${fileExtension}` const localUri = this.localStorage.getLocalUri(filename) + let pending = savingIds.get(this.powersync) + if (!pending) savingIds.set(this.powersync, (pending = new Set())) + if (pending.has(resolvedId)) { + throw new Error(`Attachment with id ${resolvedId} is already being saved`) + } + pending.add(resolvedId) + + try { + return await this.withLoadedAttachment(resolvedId, () => + this.withAttachmentContext(async (ctx) => { + // A missing in-memory row is not proof that SQLite has no attachment. + if ( + this.collection.get(resolvedId) || + (await ctx.db.getOptional( + sanitizeSQL`SELECT id FROM ${ctx.tableName} WHERE id = ?`, + [resolvedId], + )) + ) { + throw new Error(`Attachment with id ${resolvedId} already exists`) + } + + try { + const size = await this.localStorage.saveFile(localUri, data) + const attachment: AttachmentQueueRow = { + id: resolvedId, + filename, + media_type: mediaType ?? null, + local_uri: localUri, + state: AttachmentState.QUEUED_UPLOAD, + has_synced: 0, + size, + timestamp: new Date().getTime(), + meta_data: metaData ?? null, + } + + const tanStackDBTransaction = createTransaction({ + autoCommit: false, + mutationFn: async ({ transaction }) => { + await new PowerSyncTransactor({ + database: ctx.db, + }).applyTransaction(transaction) + }, + }) + + await this.runInTransaction(tanStackDBTransaction, () => { + this.collection.insert(attachment) + // allow the user to associate values in this transaction + updateHook?.(attachment) + }) + return attachment + } catch (error) { + /** + * The file is written before the transaction opens, so a failed transaction would + * otherwise leave an orphaned file behind that no attachment record points to. + */ + await this.deleteLocalFile(localUri) + throw error + } + }), + ) + } finally { + pending.delete(resolvedId) + } + } - return this.withAttachmentContext(async (ctx) => { - /** - * Checked before the file is written. Writing first would overwrite the existing - * attachment's local file, and the cleanup below would then delete it — leaving the - * pre-existing record pointing at a file that no longer exists. - * - * Deliberately outside the `try`: this throw must not reach the cleanup, because the file - * at `localUri` belongs to the pre-existing attachment rather than to this call. - */ - if (this.collection.get(resolvedId)) { - throw new Error(`Attachment with id ${resolvedId} already exists`) - } - - const size = await this.localStorage.saveFile(localUri, data) - - const attachment: AttachmentQueueRow = { - id: resolvedId, - filename, - media_type: mediaType ?? null, - local_uri: localUri, - state: AttachmentState.QUEUED_UPLOAD, - has_synced: 0, - size, - timestamp: new Date().getTime(), - meta_data: metaData ?? null, - } - - try { + /** + * Queues a file for deletion from local and remote storage. + * + * Exposes an `updateHook` option which is called inside a TanStackDB transaction, + * relational associations with the provided attachment ID should be cleaned up in this hook. + */ + async delete({ id, updateHook }: DeleteOptions): Promise { + await this.withLoadedAttachment(id, () => + this.withAttachmentContext(async (ctx) => { const tanStackDBTransaction = createTransaction({ autoCommit: false, mutationFn: async ({ transaction }) => { @@ -128,55 +182,41 @@ export class TanStackDBAttachmentQueue extends AttachmentQueue { }) await this.runInTransaction(tanStackDBTransaction, () => { - this.collection.insert(attachment) + const attachment = this.collection.get(id) + if (!attachment) { + throw new Error(`Attachment with id ${id} not found`) + } + + this.collection.update(id, (draft) => { + draft.state = AttachmentState.QUEUED_DELETE + draft.has_synced = 0 + }) + // allow the user to associate values in this transaction updateHook?.(attachment) }) - } catch (error) { - /** - * The file is written before the transaction opens, so a failed transaction would - * otherwise leave an orphaned file behind that no attachment record points to. - */ - await this.deleteLocalFile(localUri) - throw error - } - - return attachment - }) + }), + ) } - /** - * Queues a file for deletion from local and remote storage. - * - * Exposes an `updateHook` option which is called inside a TanStackDB transaction, - * relational associations with the provided attachment ID should be cleaned up in this hook. - */ - async delete({ id, updateHook }: DeleteOptions): Promise { - await this.withAttachmentContext(async (ctx) => { - const tanStackDBTransaction = createTransaction({ - autoCommit: false, - mutationFn: async ({ transaction }) => { - await new PowerSyncTransactor({ - database: ctx.db, - }).applyTransaction(transaction) - }, - }) - - await this.runInTransaction(tanStackDBTransaction, () => { - const attachment = this.collection.get(id) - if (!attachment) { - throw new Error(`Attachment with id ${id} not found`) - } - - this.collection.update(id, (draft) => { - draft.state = AttachmentState.QUEUED_DELETE - draft.has_synced = 0 - }) - - // allow the user to associate values in this transaction - updateHook?.(attachment) - }) + private async withLoadedAttachment( + id: string, + operation: () => Promise, + ): Promise { + const query = createLiveQueryCollection({ + query: (q) => + q + .from({ attachment: this.collection }) + .where(({ attachment }) => eq(attachment.id, id)), }) + try { + // Acquire just this ID before opening a mutation and retain its demand + // until PowerSync has confirmed the transaction back to the collection. + await query.preload() + return await operation() + } finally { + await query.cleanup() + } } /** diff --git a/packages/powersync-db-collection/tests/ATTACHMENT-ORACLE.md b/packages/powersync-db-collection/tests/ATTACHMENT-ORACLE.md new file mode 100644 index 0000000000..bf5a2e9c43 --- /dev/null +++ b/packages/powersync-db-collection/tests/ATTACHMENT-ORACLE.md @@ -0,0 +1,37 @@ +# Attachment lifecycle oracle + +The model records user intent (absent, present, deleted) and accepted bytes. It +does not mirror the SDK's attachment-state machine. Real SQLite transactions, +TanStack collection delivery, local files, and SDK completion writes run under +the tests. Only remote upload completion and failure are controlled. + +Each command checks owner references, SQL/collection convergence, accepted file +bytes, and cleanup of every file destination touched by save. Draining checks +remote cleanup and an idle extra tick. Rejected hooks and duplicate saves must +not change the accepted reference or bytes. + +The ordinary suite includes a committed corpus, a fixed fast-check campaign, +and a fresh random campaign. To replay a generated failure, run the matching +test with `POWERSYNC_ATTACHMENT_ORACLE_SEED` and, if supplied by fast-check, +`POWERSYNC_ATTACHMENT_ORACLE_PATH`. + +## Upstream completion limitation + +The SDK currently writes the captured upload record after remote I/O without +preserving a newer `QUEUED_DELETE`. Successful upload can lose the remote delete; +failed upload can restore an obsolete upload for retry. The ordinary generated +suite excludes only deletion while upload is in flight. This is a known gap in +supported behavior, not proof that every lifecycle is correct. + +The same model and fixture retain both desired-contract histories and a full +generated schedule in `attachments-sdk-completion.repro.ts`. Run them explicitly: + +```sh +pnpm --filter @tanstack/powersync-db-collection test:upstream-repros +``` + +These are real failing assertions, not expected failures or skipped assertions. +They are separate from the normal gate because the SDK fix is upstream. Once +completion preserves newer intent, move these histories into the normal corpus +and enable in-flight deletion in its generator. Do not weaken the oracle to +accept a detached owner with leaked remote bytes or a resumed obsolete upload. diff --git a/packages/powersync-db-collection/tests/attachments-lifecycle-fixture.ts b/packages/powersync-db-collection/tests/attachments-lifecycle-fixture.ts new file mode 100644 index 0000000000..8653e2d70c --- /dev/null +++ b/packages/powersync-db-collection/tests/attachments-lifecycle-fixture.ts @@ -0,0 +1,467 @@ +import { randomUUID } from 'node:crypto' +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { fc } from '@fast-check/vitest' +import { AttachmentTable, Schema, Table, column } from '@powersync/common' +import { NodeFileSystemAdapter, PowerSyncDatabase } from '@powersync/node' +import { createCollection } from '@tanstack/db' +import pDefer from 'p-defer' +import { expect, vi } from 'vitest' +import { powerSyncCollectionOptions } from '../src' +import { TanStackDBAttachmentQueue } from '../src/attachments' +import { TEST_DATABASE_IMPLEMENTATION } from './test-db-implementation' +import type { RemoteStorageAdapter } from '@powersync/common' + +const schema = new Schema({ + owners: new Table({ photo_id: column.text }), + attachments: new AttachmentTable(), +}) +const attachmentId = `photo` +const ownerId = `owner` + +type Command = + | `save` + | `reject-save` + | `duplicate-warm` + | `duplicate-cold` + | `delete` + | `reject-delete` + | `start-upload` + | `finish-upload` + | `fail-upload` + | `drain` + +export interface History { + name: string + commands: Array + bytes: Array +} + +// This is a user-intent model, not an AttachmentState transition table. An ID +// owns bytes only after save commits. Rejected operations leave intent intact. +// Transport completion cannot alter the most recent committed reference. +interface Model { + intent: `absent` | `present` | `deleted` + bytes: Array +} + +interface AttachmentRow { + id: string + local_uri: string | null + size: number | null + state: number + has_synced: number +} + +async function setupOracle() { + const directory = await mkdtemp(join(tmpdir(), `ps-attachment-oracle-`)) + const db = new PowerSyncDatabase({ + database: { + dbFilename: `${randomUUID()}.sqlite`, + dbLocation: directory, + implementation: TEST_DATABASE_IMPLEMENTATION, + }, + schema, + }) + await db.disconnectAndClear() + const local = new NodeFileSystemAdapter(join(directory, `files`)) + await local.initialize() + const attachments = createCollection( + powerSyncCollectionOptions({ + database: db, + table: schema.props.attachments, + }), + ) + const owners = createCollection( + powerSyncCollectionOptions({ database: db, table: schema.props.owners }), + ) + await Promise.all([attachments.stateWhenReady(), owners.stateWhenReady()]) + + // Only the remote transport is controlled. Local bytes, transactions, SQL, + // collection delivery, and SDK completion writes all use the real adapters. + const remote = new Map>() + const io = { uploads: 0, deletes: 0, downloads: 0 } + let uploadGate: + | { + entered: ReturnType> + outcome: ReturnType> + } + | undefined + let syncing: Promise | undefined + const transport: RemoteStorageAdapter = { + async uploadFile(data, attachment) { + io.uploads++ + const captured = Array.from(new Uint8Array(data)) + const gate = uploadGate + if (gate) { + gate.entered.resolve() + if (!(await gate.outcome.promise)) + throw new Error(`injected upload failure`) + } + remote.set(attachment.id, captured) + }, + deleteFile(attachment) { + io.deletes++ + remote.delete(attachment.id) + return Promise.resolve() + }, + downloadFile() { + io.downloads++ + throw new Error(`download is outside this save/delete history`) + }, + } + const queue = new TanStackDBAttachmentQueue({ + db, + attachmentsCollection: attachments, + localStorage: local, + remoteStorage: transport, + // These histories exercise explicit save/delete intent. Watcher lifecycle + // has its own boundary tests; no periodic timer can race our schedule. + watchAttachments: () => {}, + archivedCacheLimit: 0, + }) + // Track actual destinations, including writes whose transaction later fails. + // A fixed ID-derived filename would miss orphaned call-owned files. + const writtenFiles = new Set() + const saveFile = local.saveFile.bind(local) + vi.spyOn(local, `saveFile`).mockImplementation((path, data) => { + writtenFiles.add(path) + return saveFile(path, data) + }) + let uri: string | undefined + + async function finishUpload(succeeds: boolean) { + if (!uploadGate || !syncing) throw new Error(`no upload in flight`) + uploadGate.outcome.resolve(succeeds) + await syncing + syncing = undefined + uploadGate = undefined + } + + async function assertObserved(model: Model, label: string, drained: boolean) { + const rows = await db.getAll( + `SELECT id, local_uri, size, state, has_synced FROM attachments ORDER BY id`, + ) + const references = await db.getAll<{ id: string; photo_id: string | null }>( + `SELECT id, photo_id FROM owners ORDER BY id`, + ) + // SQL is authoritative; collection convergence is an additional assertion, + // never the source of the oracle's expected reference or expected bytes. + await vi.waitFor( + () => { + expect( + attachments.toArray.map( + ({ id, local_uri, size, state, has_synced }) => ({ + id, + local_uri, + size, + state, + has_synced, + }), + ), + label, + ).toEqual(rows) + expect( + owners.toArray.map(({ id, photo_id }) => ({ id, photo_id })), + label, + ).toEqual(references) + }, + { timeout: 3000, interval: 10 }, + ) + const exists = uri !== undefined && (await local.fileExists(uri)) + for (const path of writtenFiles) { + if ( + path !== uri || + model.intent === `absent` || + (drained && model.intent === `deleted`) + ) { + expect( + await local.fileExists(path), + `${label}: unexpected file ${path}`, + ).toBe(false) + } + } + const observed = { + references, + localBytes: exists + ? Array.from(new Uint8Array(await local.readFile(uri!))) + : null, + remoteBytes: remote.get(attachmentId) ?? null, + rows, + } + const evidence = `${label}; I/O=${JSON.stringify(io)}` + expect(references, label).toEqual( + model.intent === `absent` + ? [] + : [ + { + id: ownerId, + photo_id: model.intent === `present` ? attachmentId : null, + }, + ], + ) + // This is a logical FK: PowerSync's synced table views do not enforce a + // SQLite FOREIGN KEY. An attached owner must still resolve to its own row. + if (model.intent === `present`) { + expect(observed, evidence).toMatchObject({ + localBytes: model.bytes, + rows: [{ id: attachmentId, local_uri: uri, size: model.bytes.length }], + }) + if (drained) expect(observed.remoteBytes, label).toEqual(model.bytes) + } else if (model.intent === `absent` || drained) { + expect(observed, evidence).toMatchObject({ + localBytes: null, + remoteBytes: null, + rows: [], + }) + } + // If remote bytes exist at an intermediate boundary, they must come from + // the accepted save, never the rejected duplicate (same size, other bytes). + if (observed.remoteBytes) + expect(observed.remoteBytes, label).toEqual(model.bytes) + } + + async function run(command: Command, model: Model) { + switch (command) { + case `save`: + case `reject-save`: { + const saving = queue.save({ + id: attachmentId, + fileExtension: `bin`, + data: new Uint8Array(model.bytes).buffer, + updateHook: (attachment) => { + owners.insert({ id: ownerId, photo_id: attachment.id }) + if (command === `reject-save`) + throw new Error(`injected hook failure`) + }, + }) + if (command === `reject-save`) { + await expect(saving).rejects.toThrow(`injected hook failure`) + } else { + const saved = await saving + uri = saved.local_uri ?? undefined + model.intent = `present` + } + break + } + case `duplicate-warm`: + case `duplicate-cold`: { + const collection = + command === `duplicate-cold` + ? createCollection( + powerSyncCollectionOptions({ + database: db, + table: schema.props.attachments, + }), + ) + : attachments + const duplicateQueue = new TanStackDBAttachmentQueue({ + db, + attachmentsCollection: collection, + localStorage: local, + remoteStorage: transport, + watchAttachments: () => {}, + }) + try { + if (command === `duplicate-cold`) + expect(collection.get(attachmentId)).toBeUndefined() + await expect( + duplicateQueue.save({ + id: attachmentId, + fileExtension: `bin`, + data: new Uint8Array(model.bytes.map((byte) => byte ^ 255)) + .buffer, + updateHook: () => + owners.update(ownerId, (owner) => { + owner.photo_id = null + }), + }), + ).rejects.toThrow() + } finally { + await duplicateQueue.stopSync() + if (collection !== attachments) await collection.cleanup() + } + break + } + case `delete`: + case `reject-delete`: { + const deleting = queue.delete({ + id: attachmentId, + updateHook: () => { + owners.update(ownerId, (owner) => { + owner.photo_id = null + }) + if (command === `reject-delete`) + throw new Error(`injected hook failure`) + }, + }) + if (command === `reject-delete`) { + await expect(deleting).rejects.toThrow(`injected hook failure`) + } else { + await deleting + model.intent = `deleted` + } + break + } + case `start-upload`: + if (syncing) throw new Error(`upload already in flight`) + uploadGate = { entered: pDefer(), outcome: pDefer() } + syncing = queue.syncStorage() + await vi.waitFor(() => expect(io.uploads).toBeGreaterThan(0), { + timeout: 3000, + interval: 10, + }) + await uploadGate.entered.promise + break + case `finish-upload`: + await finishUpload(true) + break + case `fail-upload`: + await finishUpload(false) + break + case `drain`: { + if (syncing) throw new Error(`finish the held upload before draining`) + // No more injected failures: two passes suffice for these single-ID + // histories. A third tick must be idle, not merely leave SQL detached. + await queue.syncStorage() + await queue.syncStorage() + const settledIO = { ...io } + await queue.syncStorage() + expect(io).toEqual(settledIO) + break + } + } + } + + return { + run, + assertObserved, + async dispose() { + if (uploadGate) uploadGate.outcome.resolve(true) + await syncing + await queue.stopSync() + await Promise.all([attachments.cleanup(), owners.cleanup()]) + await db.disconnectAndClear() + await db.close() + await local.clear() + await rm(directory, { recursive: true }) + }, + } +} + +export async function runHistory(history: History) { + const fixture = await setupOracle() + const model: Model = { intent: `absent`, bytes: [...history.bytes] } + try { + for (const [index, command] of history.commands.entries()) { + const label = `${history.name}: step ${index + 1}/${history.commands.length} ${command}; ${JSON.stringify(history)}` + await fixture.run(command, model) + await fixture.assertObserved(model, label, command === `drain`) + } + } finally { + await fixture.dispose() + } +} + +export const corpus: Array> = [ + { + name: `warm duplicate preserves bytes and reference`, + commands: [`save`, `duplicate-warm`, `drain`], + }, + { + name: `cold duplicate preserves bytes and reference`, + commands: [`save`, `duplicate-cold`, `drain`], + }, + { + name: `delete before upload removes local data`, + commands: [`save`, `delete`, `drain`], + }, + { + name: `delete during successful upload removes remote data`, + commands: [`save`, `start-upload`, `delete`, `finish-upload`, `drain`], + }, + { + name: `delete after upload removes remote data`, + commands: [`save`, `start-upload`, `finish-upload`, `delete`, `drain`], + }, + { + name: `delete during failed upload does not retry obsolete intent`, + commands: [`save`, `start-upload`, `delete`, `fail-upload`, `drain`], + }, + { + name: `upload failure retries accepted bytes`, + commands: [`save`, `start-upload`, `fail-upload`, `drain`], + }, + { + name: `delete after upload failure removes local data`, + commands: [`save`, `start-upload`, `fail-upload`, `delete`, `drain`], + }, + { + name: `save hook failure rolls back both rows and local data`, + commands: [`reject-save`, `save`, `drain`], + }, + { + name: `delete hook failure preserves bytes and reference`, + commands: [`save`, `reject-delete`, `drain`], + }, + { + name: `repeated delete stays deleted`, + commands: [`save`, `delete`, `delete`, `drain`], + }, +] + +// The ordinary suite excludes only the named SDK completion race; the opt-in +// repro suite runs that same law and harness without an expected-failure waiver. +export const sdkCompletionCorpus = corpus.filter( + ({ commands }) => + commands.indexOf(`delete`) > commands.indexOf(`start-upload`) && + commands.indexOf(`delete`) < + Math.max( + commands.indexOf(`finish-upload`), + commands.indexOf(`fail-upload`), + ), +) +export const supportedCorpus = corpus.filter( + (history) => !sdkCompletionCorpus.includes(history), +) + +export function historyArbitrary(includeInFlightDelete: boolean) { + return fc + .record({ + bytes: fc.array(fc.integer({ min: 0, max: 255 }), { + minLength: 1, + maxLength: 16, + }), + duplicate: fc.constantFrom(`none`, `warm`, `cold`), + deletion: includeInFlightDelete + ? fc.constantFrom(`never`, `before`, `during`, `after`) + : fc.constantFrom(`never`, `before`, `after`), + uploadSucceeds: fc.boolean(), + rejectDelete: fc.boolean(), + }) + .map( + ({ + bytes, + duplicate, + deletion, + uploadSucceeds, + rejectDelete, + }): History => { + const commands: Array = [`save`] + if (duplicate === `warm`) commands.push(`duplicate-warm`) + if (duplicate === `cold`) commands.push(`duplicate-cold`) + if (rejectDelete) commands.push(`reject-delete`) + if (deletion === `before`) { + commands.push(`delete`) + } else { + commands.push(`start-upload`) + if (deletion === `during`) commands.push(`delete`) + commands.push(uploadSucceeds ? `finish-upload` : `fail-upload`) + if (deletion === `after`) commands.push(`delete`) + } + commands.push(`drain`) + return { name: `generated`, commands, bytes } + }, + ) +} diff --git a/packages/powersync-db-collection/tests/attachments-lifecycle-oracle.test.ts b/packages/powersync-db-collection/tests/attachments-lifecycle-oracle.test.ts new file mode 100644 index 0000000000..2b5e50d448 --- /dev/null +++ b/packages/powersync-db-collection/tests/attachments-lifecycle-oracle.test.ts @@ -0,0 +1,37 @@ +import { fc } from '@fast-check/vitest' +import { describe, it } from 'vitest' +import { + historyArbitrary, + runHistory, + supportedCorpus, +} from './attachments-lifecycle-fixture' +import { TEST_DATABASE_IMPLEMENTATION } from './test-db-implementation' + +const describePowerSync = TEST_DATABASE_IMPLEMENTATION + ? describe + : describe.skip + +describePowerSync(`attachment lifecycle intent oracle`, () => { + it.each(supportedCorpus)(`$name`, async (history) => { + await runHistory({ ...history, bytes: [0, 7, 128, 255] }) + }) + + it.each([`fixed`, `random`] as const)( + `preserves intent across %s command histories`, + async (campaign) => { + const replaySeed = process.env.POWERSYNC_ATTACHMENT_ORACLE_SEED + await fc.assert(fc.asyncProperty(historyArbitrary(false), runHistory), { + ...(replaySeed + ? { seed: Number(replaySeed) } + : campaign === `fixed` + ? { seed: 1616 } + : {}), + numRuns: 12, + ...(process.env.POWERSYNC_ATTACHMENT_ORACLE_PATH + ? { path: process.env.POWERSYNC_ATTACHMENT_ORACLE_PATH } + : {}), + }) + }, + 30000, + ) +}) diff --git a/packages/powersync-db-collection/tests/attachments-native-sdk.repro.ts b/packages/powersync-db-collection/tests/attachments-native-sdk.repro.ts new file mode 100644 index 0000000000..81920c1b84 --- /dev/null +++ b/packages/powersync-db-collection/tests/attachments-native-sdk.repro.ts @@ -0,0 +1,85 @@ +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { + AttachmentQueue, + AttachmentState, + AttachmentTable, + Schema, +} from '@powersync/common' +import { NodeFileSystemAdapter, PowerSyncDatabase } from '@powersync/node' +import pDefer from 'p-defer' +import { describe, expect, it } from 'vitest' +import { TEST_DATABASE_IMPLEMENTATION } from './test-db-implementation' + +const describePowerSync = TEST_DATABASE_IMPLEMENTATION + ? describe + : describe.skip + +// No TanStack collection, subclass, transaction, or watcher participates in this +// control. The native SDK alone loses deletion intent in its completion write. +describePowerSync(`native SDK completion contract`, () => { + it.each([true, false])( + `keeps deletion queued after upload success=%s`, + async (succeeds) => { + const directory = await mkdtemp(join(tmpdir(), `ps-native-completion-`)) + const db = new PowerSyncDatabase({ + database: { + dbFilename: `attachments.sqlite`, + dbLocation: directory, + implementation: TEST_DATABASE_IMPLEMENTATION, + }, + schema: new Schema({ attachments: new AttachmentTable() }), + }) + await db.disconnectAndClear() + const localStorage = new NodeFileSystemAdapter(join(directory, `files`)) + await localStorage.initialize() + const entered = pDefer() + const release = pDefer() + const queue = new AttachmentQueue({ + db, + localStorage, + remoteStorage: { + async uploadFile() { + entered.resolve() + await release.promise + if (!succeeds) throw new Error(`injected upload failure`) + }, + async deleteFile() {}, + downloadFile() { + throw new Error(`unexpected download`) + }, + }, + watchAttachments: () => {}, + }) + let syncing: Promise | undefined + try { + const record = await queue.saveFile({ + data: new Uint8Array([0, 7, 128, 255]).buffer, + fileExtension: `bin`, + }) + syncing = queue.syncStorage() + await entered.promise + await queue.deleteFile({ id: record.id }) + const readState = () => + db.get(`SELECT state FROM attachments WHERE id = ?`, [record.id]) + expect(await readState()).toEqual({ + state: AttachmentState.QUEUED_DELETE, + }) + release.resolve() + await syncing + expect(await readState()).toEqual({ + state: AttachmentState.QUEUED_DELETE, + }) + } finally { + release.resolve() + await syncing + await queue.stopSync() + await db.disconnectAndClear() + await db.close() + await localStorage.clear() + await rm(directory, { recursive: true }) + } + }, + ) +}) diff --git a/packages/powersync-db-collection/tests/attachments-sdk-completion.repro.ts b/packages/powersync-db-collection/tests/attachments-sdk-completion.repro.ts new file mode 100644 index 0000000000..0026ab9437 --- /dev/null +++ b/packages/powersync-db-collection/tests/attachments-sdk-completion.repro.ts @@ -0,0 +1,30 @@ +import { fc } from '@fast-check/vitest' +import { describe, it } from 'vitest' +import { + historyArbitrary, + runHistory, + sdkCompletionCorpus, +} from './attachments-lifecycle-fixture' +import { TEST_DATABASE_IMPLEMENTATION } from './test-db-implementation' + +const describePowerSync = TEST_DATABASE_IMPLEMENTATION + ? describe + : describe.skip + +// Desired contract, deliberately not test.fails: run separately until the SDK +// preserves newer deletion intent in both successful and failed upload writes. +describePowerSync(`upstream attachment completion contract`, () => { + it.each(sdkCompletionCorpus)(`$name`, async (history) => { + await runHistory({ ...history, bytes: [0, 7, 128, 255] }) + }) + + it(`preserves deletion across all generated completion schedules`, async () => { + await fc.assert(fc.asyncProperty(historyArbitrary(true), runHistory), { + seed: Number(process.env.POWERSYNC_ATTACHMENT_ORACLE_SEED ?? 1616), + numRuns: 12, + ...(process.env.POWERSYNC_ATTACHMENT_ORACLE_PATH + ? { path: process.env.POWERSYNC_ATTACHMENT_ORACLE_PATH } + : {}), + }) + }, 30000) +}) diff --git a/packages/powersync-db-collection/tests/attachments.test.ts b/packages/powersync-db-collection/tests/attachments.test.ts index 193f8fa0c1..1cafe169fd 100644 --- a/packages/powersync-db-collection/tests/attachments.test.ts +++ b/packages/powersync-db-collection/tests/attachments.test.ts @@ -1,6 +1,7 @@ import { randomUUID } from 'node:crypto' import { tmpdir } from 'node:os' import { join } from 'node:path' +import pDefer from 'p-defer' import { AttachmentState, AttachmentTable, @@ -56,7 +57,7 @@ const describePowerSync = TEST_DATABASE_IMPLEMENTATION : describe.skip describePowerSync(`PowerSync AttachmentQueue (TanStackDB)`, () => { - async function setup() { + async function setup(syncMode: `eager` | `on-demand` = `eager`) { const db = new PowerSyncDatabase({ database: { dbFilename: `attachments-test-${randomUUID()}.sqlite`, @@ -91,6 +92,7 @@ describePowerSync(`PowerSync AttachmentQueue (TanStackDB)`, () => { powerSyncCollectionOptions({ database: db, table: APP_SCHEMA.props.attachments, + syncMode, }), ) const usersCollection = createCollection( @@ -212,6 +214,292 @@ describePowerSync(`PowerSync AttachmentQueue (TanStackDB)`, () => { } describe(`save`, () => { + it(`serializes successive watched snapshots through the SDK context`, async () => { + const fixture = await setup() + let update!: Parameters[0] + const queue = fixture.createQueue({ + archivedCacheLimit: 100, + watchAttachments: (callback) => { + update = callback + }, + }) + const record = await queue.save({ + data: createMockJpegBuffer(), + fileExtension: `jpg`, + }) + await queue.syncStorage() + await queue.startSync() + const entered = pDefer() + const release = pDefer() + const held = queue.withAttachmentContext(async () => { + entered.resolve() + await release.promise + }) + await entered.promise + const completed: Array = [] + const first = update([]).then(() => { + completed.push(1) + }) + const second = update([{ id: record.id, fileExtension: `jpg` }]).then( + () => { + completed.push(2) + }, + ) + try { + await Promise.resolve() + expect(completed).toEqual([]) + release.resolve() + await Promise.all([held, first, second]) + expect(completed).toEqual([1, 2]) + expect( + await fixture.db.get(`SELECT state FROM attachments WHERE id = ?`, [ + record.id, + ]), + ).toEqual({ state: AttachmentState.SYNCED }) + } finally { + release.resolve() + await Promise.all([held, first, second]) + } + }) + + it(`preserves the winner file across two queue instances sharing storage`, async () => { + const fixture = await setup() + const first = fixture.createQueue() + const second = fixture.createQueue() + const entered = pDefer() + const release = pDefer() + const saveFile = fixture.localStorage.saveFile.bind(fixture.localStorage) + const write = vi + .spyOn(fixture.localStorage, `saveFile`) + .mockImplementation(async (...args) => { + const size = await saveFile(...args) + if (write.mock.calls.length === 1) { + entered.resolve() + await release.promise + } + return size + }) + const saving = first.save({ + id: `shared-id`, + data: createMockJpegBuffer(), + fileExtension: `jpg`, + }) + try { + await entered.promise + await expect( + second.save({ + id: `shared-id`, + data: new Uint8Array([7, 8, 9]).buffer, + fileExtension: `jpg`, + }), + ).rejects.toThrow(/already/) + expect(write).toHaveBeenCalledTimes(1) + } finally { + release.resolve() + await saving + } + const winner = await saving + expect( + new Uint8Array(await fixture.localStorage.readFile(winner.local_uri!)), + ).toEqual(new Uint8Array(createMockJpegBuffer())) + }) + + it.each([`ready`, `cold`, `on-demand`] as const)( + `preserves an existing file when reused through a %s collection`, + async (phase) => { + const fixture = await setup() + const original = await fixture.createQueue().save({ + id: `reused-after-reopen`, + data: createMockJpegBuffer(), + fileExtension: `jpg`, + }) + const collection = + phase === `ready` + ? fixture.attachmentsCollection + : createCollection( + powerSyncCollectionOptions({ + database: fixture.db, + table: APP_SCHEMA.props.attachments, + syncMode: phase === `on-demand` ? `on-demand` : `eager`, + }), + ) + onTestFinished(() => collection.cleanup()) + const queue = new TanStackDBAttachmentQueue({ + db: fixture.db, + attachmentsCollection: collection, + localStorage: fixture.localStorage, + remoteStorage: fixture.remoteStorage, + watchAttachments: () => {}, + }) + onTestFinished(() => queue.stopSync()) + if (phase !== `ready`) + expect(collection.get(original.id)).toBeUndefined() + await expect( + queue.save({ + id: original.id, + data: new Uint8Array([7, 8, 9]).buffer, + fileExtension: `jpg`, + }), + ).rejects.toThrow() + expect( + await fixture.db.getOptional( + `SELECT id FROM attachments WHERE id = ?`, + [original.id], + ), + ).toEqual({ id: original.id }) + expect(await fixture.localStorage.fileExists(original.local_uri!)).toBe( + true, + ) + expect( + new Uint8Array( + await fixture.localStorage.readFile(original.local_uri!), + ), + ).toEqual(new Uint8Array(createMockJpegBuffer())) + }, + ) + + it(`cleans a partial write and releases its ID for retry`, async () => { + const fixture = await setup() + const write = fixture.localStorage.saveFile.bind(fixture.localStorage) + const savedPaths: Array = [] + vi.spyOn(fixture.localStorage, `saveFile`).mockImplementationOnce( + async (uri, data) => { + savedPaths.push(uri) + await write(uri, data) + throw new Error(`partial write failure`) + }, + ) + const options = { + id: `partial`, + data: createMockJpegBuffer(), + fileExtension: `jpg`, + } + await expect(fixture.createQueue().save(options)).rejects.toThrow( + `partial write failure`, + ) + expect(savedPaths).toHaveLength(1) + expect(await fixture.localStorage.fileExists(savedPaths[0]!)).toBe(false) + expect( + await fixture.db.getOptional( + `SELECT id FROM attachments WHERE id = ?`, + [options.id], + ), + ).toBeNull() + const saved = await fixture.createQueue().save(options) + expect(await fixture.localStorage.fileExists(saved.local_uri!)).toBe(true) + }) + + it(`allows another queue to save a distinct ID while a write is held`, async () => { + const fixture = await setup() + const entered = pDefer() + const release = pDefer() + const write = fixture.localStorage.saveFile.bind(fixture.localStorage) + vi.spyOn(fixture.localStorage, `saveFile`).mockImplementationOnce( + async (...args) => { + const size = await write(...args) + entered.resolve() + await release.promise + return size + }, + ) + const saving = fixture + .createQueue() + .save({ + id: `held`, + data: createMockJpegBuffer(), + fileExtension: `jpg`, + }) + try { + await entered.promise + const other = await fixture + .createQueue() + .save({ + id: `other`, + data: createMockJpegBuffer(), + fileExtension: `jpg`, + }) + expect(await fixture.localStorage.fileExists(other.local_uri!)).toBe( + true, + ) + } finally { + release.resolve() + await saving + } + }) + + it.each([`before`, `after`] as const)( + `preserves delete intent %s an SDK upload`, + async (timing) => { + const fixture = await setup() + const queue = fixture.createQueue() + const uploaded = pDefer() + const release = pDefer() + const remoteFiles = new Set() + fixture.uploadFile.mockImplementation(async (_, attachment) => { + uploaded.resolve() + await release.promise + remoteFiles.add(attachment.id) + }) + fixture.deleteFile.mockImplementation((attachment) => { + remoteFiles.delete(attachment.id) + return Promise.resolve() + }) + const userId = randomUUID() + const record = await queue.save({ + data: createMockJpegBuffer(), + fileExtension: `jpg`, + updateHook: (attachment) => { + fixture.usersCollection.insert({ + id: userId, + name: `owner`, + email: null, + photo_id: attachment.id, + }) + }, + }) + let sync: Promise | undefined + try { + if (timing !== `before`) { + sync = queue.syncStorage() + await uploaded.promise + release.resolve() + await sync + } + await queue.delete({ + id: record.id, + updateHook: () => { + fixture.usersCollection.update(userId, (row) => { + row.photo_id = null + }) + }, + }) + expect( + await fixture.db.get(`SELECT state FROM attachments WHERE id = ?`, [ + record.id, + ]), + ).toEqual({ state: AttachmentState.QUEUED_DELETE }) + expect( + await fixture.db.get(`SELECT photo_id FROM users WHERE id = ?`, [ + userId, + ]), + ).toEqual({ photo_id: null }) + release.resolve() + await sync + // Complete two real SDK passes, not a timeout waiting for a mock state. + await queue.syncStorage() + await queue.syncStorage() + expect(fixture.deleteFile).toHaveBeenCalledTimes(1) + expect(remoteFiles.has(record.id)).toBe(false) + expect(await fixture.localStorage.fileExists(record.local_uri!)).toBe( + false, + ) + } finally { + release.resolve() + await sync + } + }, + ) + it(`writes the local file and inserts a QUEUED_UPLOAD row into the collection`, async () => { const { createQueue, attachmentsCollection, localStorage } = await setup() const queue = createQueue() @@ -376,7 +664,9 @@ describePowerSync(`PowerSync AttachmentQueue (TanStackDB)`, () => { expect(fulfilled).toHaveLength(1) expect(rejected).toHaveLength(1) expect(rejected[0]!.reason).toEqual( - expect.objectContaining({ message: expect.stringMatching(/exists/) }), + expect.objectContaining({ + message: expect.stringMatching(/exists|being saved/), + }), ) const winner = fulfilled[0]!.value @@ -398,9 +688,8 @@ describePowerSync(`PowerSync AttachmentQueue (TanStackDB)`, () => { } = await setup() const queue = createQueue() - // A caller-supplied id lets us derive the local uri without a returned record. const id = randomUUID() - const localUri = localStorage.getLocalUri(`${id}.jpg`) + let localUri: string | undefined await expect( queue.save({ @@ -408,6 +697,7 @@ describePowerSync(`PowerSync AttachmentQueue (TanStackDB)`, () => { data: createMockJpegBuffer(), fileExtension: `jpg`, updateHook: (attachment) => { + localUri = attachment.local_uri! usersCollection.insert({ id: randomUUID(), name: `steven`, @@ -420,7 +710,8 @@ describePowerSync(`PowerSync AttachmentQueue (TanStackDB)`, () => { ).rejects.toThrow(/updateHook failed/) // The file is written before the transaction opens, so it must be cleaned up. - expect(await localStorage.fileExists(localUri)).toBe(false) + expect(localUri).toBeDefined() + expect(await localStorage.fileExists(localUri!)).toBe(false) // Neither the attachment nor the hook's own mutation may survive the failure. expect(attachmentsCollection.get(id)).toBeUndefined() @@ -429,6 +720,110 @@ describePowerSync(`PowerSync AttachmentQueue (TanStackDB)`, () => { }) describe(`delete file`, () => { + it.each([`eager`, `on-demand`] as const)( + `can retry a failed save and later delete in %s mode`, + async (syncMode) => { + const fixture = await setup(syncMode) + const first = fixture.createQueue() + const options = { + id: `retry`, + data: createMockJpegBuffer(), + fileExtension: `jpg`, + } + await expect( + first.save({ + ...options, + updateHook: () => { + throw new Error(`hook failure`) + }, + }), + ).rejects.toThrow(`hook failure`) + const second = fixture.createQueue() + const saved = await second.save(options) + expect( + new Uint8Array(await fixture.localStorage.readFile(saved.local_uri!)), + ).toEqual(new Uint8Array(createMockJpegBuffer())) + await first.delete({ id: saved.id }) + expect( + await fixture.db.get(`SELECT state FROM attachments WHERE id = ?`, [ + saved.id, + ]), + ).toEqual({ state: AttachmentState.QUEUED_DELETE }) + }, + ) + + it(`finds a queued file after its storage root moves`, async () => { + const fixture = await setup() + const original = await fixture.createQueue().save({ + id: `moving`, + data: createMockJpegBuffer(), + fileExtension: `jpg`, + }) + const moved = new NodeFileSystemAdapter( + join(tmpdir(), `ps-moved-${randomUUID()}`), + ) + await moved.initialize() + onTestFinished(() => moved.clear()) + // Move the bytes without changing SQLite, as a changed app directory does. + const movedUri = moved.getLocalUri(original.local_uri!.split(`/`).at(-1)!) + await moved.saveFile( + movedUri, + await fixture.localStorage.readFile(original.local_uri!), + ) + await fixture.localStorage.deleteFile(original.local_uri!) + const queue = new TanStackDBAttachmentQueue({ + db: fixture.db, + attachmentsCollection: fixture.attachmentsCollection, + localStorage: moved, + remoteStorage: fixture.remoteStorage, + watchAttachments: () => {}, + }) + onTestFinished(() => queue.stopSync()) + await queue.startSync() + await vi.waitFor(() => + expect(fixture.uploadFile).toHaveBeenCalledTimes(1), + ) + expect(fixture.uploadFile.mock.calls[0]![1].localUri).toBe(movedUri) + expect(new Uint8Array(await moved.readFile(movedUri))).toEqual( + new Uint8Array(createMockJpegBuffer()), + ) + }) + + it.each([`eager`, `on-demand`] as const)( + `loads an uncached attachment before deleting in %s mode`, + async (syncMode) => { + const fixture = await setup() + const original = await fixture.createQueue().save({ + id: `uncached`, + data: createMockJpegBuffer(), + fileExtension: `jpg`, + }) + const collection = createCollection( + powerSyncCollectionOptions({ + database: fixture.db, + table: APP_SCHEMA.props.attachments, + syncMode, + }), + ) + onTestFinished(() => collection.cleanup()) + const queue = new TanStackDBAttachmentQueue({ + db: fixture.db, + attachmentsCollection: collection, + localStorage: fixture.localStorage, + remoteStorage: fixture.remoteStorage, + watchAttachments: () => {}, + }) + onTestFinished(() => queue.stopSync()) + expect(collection.get(original.id)).toBeUndefined() + await queue.delete({ id: original.id }) + expect( + await fixture.db.get(`SELECT state FROM attachments WHERE id = ?`, [ + original.id, + ]), + ).toEqual({ state: AttachmentState.QUEUED_DELETE }) + }, + ) + it(`queues an existing attachment for deletion and removes the local file`, async () => { const { createQueue, diff --git a/packages/powersync-db-collection/tests/upstream.config.ts b/packages/powersync-db-collection/tests/upstream.config.ts new file mode 100644 index 0000000000..2dd49b2631 --- /dev/null +++ b/packages/powersync-db-collection/tests/upstream.config.ts @@ -0,0 +1,12 @@ +import { defineConfig } from 'vitest/config' + +export default defineConfig({ + test: { + name: `powersync-upstream-contract-repros`, + dir: `./tests`, + include: [`**/*.repro.ts`], + environment: `node`, + coverage: { enabled: false }, + typecheck: { enabled: false }, + }, +}) From d34d2d4904f921058fd6869da365f682551d2877 Mon Sep 17 00:00:00 2001 From: Kyle Mathews Date: Thu, 10 Sep 2026 21:40:13 -0600 Subject: [PATCH 13/13] chore: add changeset for attachment ownership fixes --- .changeset/powersync-attachment-startup-ownership.md | 5 +++++ 1 file changed, 5 insertions(+) create mode 100644 .changeset/powersync-attachment-startup-ownership.md diff --git a/.changeset/powersync-attachment-startup-ownership.md b/.changeset/powersync-attachment-startup-ownership.md new file mode 100644 index 0000000000..d30adce3d6 --- /dev/null +++ b/.changeset/powersync-attachment-startup-ownership.md @@ -0,0 +1,5 @@ +--- +'@tanstack/powersync-db-collection': patch +--- + +Load attachment IDs before save/delete in eager and on-demand collections. Preserve existing files when a duplicate save is rejected, reject overlapping saves of the same ID across queues sharing a database, and clean up partial local writes.