diff --git a/.changeset/curly-planets-lead.md b/.changeset/curly-planets-lead.md new file mode 100644 index 000000000..8f0b2cbeb --- /dev/null +++ b/.changeset/curly-planets-lead.md @@ -0,0 +1,6 @@ +--- +'@tanstack/powersync-db-collection': minor +--- + +Add attachments support via `TanStackDBAttachmentQueue`. This extends the PowerSync SDK's `AttachmentQueue` and backs it with +a TanStack DB collection, so attachment metadata and related rows commit atomically. Local files and remote uploads/deletes are managed separately. diff --git a/.changeset/powersync-attachment-startup-ownership.md b/.changeset/powersync-attachment-startup-ownership.md new file mode 100644 index 000000000..d30adce3d --- /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. diff --git a/docs/collections/powersync-collection.md b/docs/collections/powersync-collection.md index 3860e5495..b2a1ff852 100644 --- a/docs/collections/powersync-collection.md +++ b/docs/collections/powersync-collection.md @@ -1097,4 +1097,177 @@ const liveQuery = createLiveQueryCollection({ completed: todo.completed, })), }) -``` \ No newline at end of file +``` + +## Attachments + +`@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. + +> 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. + +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" + +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, + }) +} +``` + +### 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. + +```ts +await attachmentQueue.save({ + data, // file bytes (ArrayBuffer / base64, per your local adapter) + fileExtension: "jpg", + updateHook: (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 + }) + }, +}) +``` + +> `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. 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, + updateHook: () => { + 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, + })) +) +``` diff --git a/packages/powersync-db-collection/package.json b/packages/powersync-db-collection/package.json index cee880602..975091723 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", @@ -59,11 +60,11 @@ "p-defer": "^4.0.1" }, "peerDependencies": { - "@powersync/common": "^1.41.0" + "@powersync/common": "^1.57.0" }, "devDependencies": { - "@powersync/common": "1.49.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 new file mode 100644 index 000000000..465d45d13 --- /dev/null +++ b/packages/powersync-db-collection/src/attachments.ts @@ -0,0 +1,263 @@ +import { + AttachmentQueue, + AttachmentState, + sanitizeSQL, +} from '@powersync/common' +import { createLiveQueryCollection, createTransaction, eq } from '@tanstack/db' +import { PowerSyncTransactor } from './PowerSyncTransactor' + +import type { + AbstractPowerSyncDatabase, + AttachmentData, + AttachmentQueueOptions, + AttachmentTable, +} from '@powersync/common' +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. + * In order to have the same relational data be set in a single transaction. + * This also allows for joining both TanStackDB collections. + */ + attachmentsCollection: Collection +} + +export interface SaveOptions { + data: AttachmentData + fileExtension: string + mediaType?: string + metaData?: string + /** + * Optional custom ID. If not provided, a UUID will be generated. + * + * Rejected if this ID is already in the queue or is being saved by another + * call sharing the same PowerSync database object. + */ + id?: string + /** + * 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 +} + +export interface DeleteOptions { + id: string + /** + * 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 +} + +export type AttachmentQueueRow = OptionalExtractedTable + +/** + * 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 save({ + data, + fileExtension, + mediaType, + metaData, + id, + updateHook, + }: SaveOptions): Promise { + 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) + } + } + + /** + * 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 }) => { + 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() + } + } + + /** + * 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/src/index.ts b/packages/powersync-db-collection/src/index.ts index f8d092805..f96a7a0ee 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/ATTACHMENT-ORACLE.md b/packages/powersync-db-collection/tests/ATTACHMENT-ORACLE.md new file mode 100644 index 000000000..bf5a2e9c4 --- /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 000000000..8653e2d70 --- /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 000000000..2b5e50d44 --- /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 000000000..81920c1b8 --- /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 000000000..0026ab943 --- /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 new file mode 100644 index 000000000..1cafe169f --- /dev/null +++ b/packages/powersync-db-collection/tests/attachments.test.ts @@ -0,0 +1,952 @@ +import { randomUUID } from 'node:crypto' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import pDefer from 'p-defer' +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' +import type { AttachmentQueueRow } from '../src/attachments' + +// 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(syncMode: `eager` | `on-demand` = `eager`) { + 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, + syncMode, + }), + ) + 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(`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() + + const data = new Uint8Array(123).fill(42).buffer + const record = await queue.save({ + 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.save({ + data: createMockJpegBuffer(), + fileExtension: `jpg`, + updateHook: (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.save({ + data: createMockJpegBuffer(), + fileExtension: `jpg`, + updateHook: (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.save({ + id, + data: createMockJpegBuffer(), + fileExtension: `png`, + }) + + expect(record.id).toBe(id) + 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(`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|being saved/), + }), + ) + + 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, + attachmentsCollection, + usersCollection, + localStorage, + } = await setup() + const queue = createQueue() + + const id = randomUUID() + let localUri: string | undefined + + await expect( + queue.save({ + id, + data: createMockJpegBuffer(), + fileExtension: `jpg`, + updateHook: (attachment) => { + localUri = attachment.local_uri! + 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(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() + expect(usersCollection.size).toBe(0) + }) + }) + + 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, + attachmentsCollection, + usersCollection, + localStorage, + } = await setup() + const queue = createQueue() + await queue.startSync() + + 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, + }) + }, + }) + + await waitForState( + attachmentsCollection, + record.id, + AttachmentState.SYNCED, + ) + + await queue.delete({ + id: record.id, + updateHook: (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.delete({ 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) + }) + + 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) + }) + }) +}) 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 000000000..2dd49b263 --- /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 }, + }, +}) diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index d52d43d8f..c9512eaf0 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1418,11 +1418,11 @@ importers: version: 4.0.1 devDependencies: '@powersync/common': - specifier: 1.49.0 - version: 1.49.0 + specifier: 1.57.0 + version: 1.57.0 '@powersync/node': - specifier: 0.18.1 - version: 0.18.1(@powersync/common@1.49.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 @@ -5156,13 +5156,13 @@ 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.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: @@ -7127,9 +7127,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==} @@ -8501,9 +8498,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'} @@ -9705,6 +9699,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==} @@ -17019,16 +17016,13 @@ snapshots: '@poppinss/exception@1.2.3': {} - '@powersync/common@1.49.0': + '@powersync/common@1.57.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.19.2(@powersync/common@1.57.0)(better-sqlite3@12.8.0)': dependencies: - '@powersync/common': 1.49.0 - async-mutex: 0.5.0 - bson: 6.10.4 + '@powersync/common': 1.57.0 comlink: 4.4.2 undici: 7.24.4 optionalDependencies: @@ -19500,10 +19494,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: {} @@ -21121,8 +21111,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 @@ -22680,6 +22668,8 @@ snapshots: js-base64@3.7.8: {} + js-logger@1.6.1: {} + js-tokens@4.0.0: {} js-tokens@9.0.1: {}