-
Notifications
You must be signed in to change notification settings - Fork 26
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
wip: split worker db interfaces and instantiation
- Loading branch information
1 parent
e937099
commit e06f3a9
Showing
23 changed files
with
854 additions
and
643 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
269 changes: 269 additions & 0 deletions
269
packages/web/src/db/adapters/LockedAsyncDatabaseAdapter.ts
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,269 @@ | ||
| import { | ||
| BaseObserver, | ||
| DBAdapter, | ||
| DBAdapterListener, | ||
| DBGetUtils, | ||
| DBLockOptions, | ||
| LockContext, | ||
| QueryResult, | ||
| Transaction | ||
| } from '@powersync/common'; | ||
| import Logger, { ILogger } from 'js-logger'; | ||
| import { getNavigatorLocks } from '../..//shared/navigator'; | ||
| import { AsyncDatabaseConnection } from './AsyncDatabaseConnection'; | ||
|
|
||
| /** | ||
| * @internal | ||
| */ | ||
| export interface LockedAsyncDatabaseAdapterOptions { | ||
| name: string; | ||
| openConnection: () => Promise<AsyncDatabaseConnection>; | ||
| debugMode?: boolean; | ||
| logger?: ILogger; | ||
| } | ||
|
|
||
| type LockedAsyncDatabaseAdapterListener = DBAdapterListener & { | ||
| initialized?: () => void; | ||
| }; | ||
|
|
||
| /** | ||
| * @internal | ||
| * Wraps a {@link AsyncDatabaseConnection} and provides exclusive locking functions in | ||
| * order to implement {@link DBAdapter}. | ||
| */ | ||
| export class LockedAsyncDatabaseAdapter extends BaseObserver<LockedAsyncDatabaseAdapterListener> implements DBAdapter { | ||
| private logger: ILogger; | ||
| private dbGetHelpers: DBGetUtils | null; | ||
| private debugMode: boolean; | ||
| private _dbIdentifier: string; | ||
| private _isInitialized = false; | ||
| private _db: AsyncDatabaseConnection | null = null; | ||
| private _disposeTableChangeListener: (() => void) | null = null; | ||
|
|
||
| constructor(protected options: LockedAsyncDatabaseAdapterOptions) { | ||
| super(); | ||
| this._dbIdentifier = options.name; | ||
| this.logger = options.logger ?? Logger.get(`LockedAsyncDatabaseAdapter - ${this._dbIdentifier}`); | ||
| // Set the name if provided. We can query for the name if not available yet | ||
| this.debugMode = options.debugMode ?? false; | ||
| if (this.debugMode) { | ||
| const originalExecute = this._execute.bind(this); | ||
| this._execute = async (sql, bindings) => { | ||
| const start = performance.now(); | ||
| try { | ||
| const r = await originalExecute(sql, bindings); | ||
| performance.measure(`[SQL] ${sql}`, { start }); | ||
| return r; | ||
| } catch (e: any) { | ||
| performance.measure(`[SQL] [ERROR: ${e.message}] ${sql}`, { start }); | ||
| throw e; | ||
| } | ||
| }; | ||
| } | ||
|
|
||
| this.dbGetHelpers = this.generateDBHelpers({ | ||
| execute: (query, params) => this.acquireLock(() => this._execute(query, params)) | ||
| }); | ||
| } | ||
|
|
||
| protected get baseDB() { | ||
| if (!this._db) { | ||
| throw new Error(`Initialization has not completed yet. Cannot access base db`); | ||
| } | ||
| return this._db; | ||
| } | ||
|
|
||
| get name() { | ||
| return this._dbIdentifier; | ||
| } | ||
|
|
||
| async init() { | ||
| this._db = await this.options.openConnection(); | ||
| await this._db.init(); | ||
| this._disposeTableChangeListener = await this._db.registerOnTableChange((event) => { | ||
| this.iterateListeners((cb) => cb.tablesUpdated?.(event)); | ||
| }); | ||
| this._isInitialized = true; | ||
| this.iterateListeners((cb) => cb.initialized?.()); | ||
| } | ||
|
|
||
| protected async waitForInitialized() { | ||
| if (this._isInitialized) { | ||
| return; | ||
| } | ||
| return new Promise<void>((resolve) => { | ||
| const l = this.registerListener({ | ||
| initialized: () => { | ||
| resolve(); | ||
| l(); | ||
| } | ||
| }); | ||
| }); | ||
| } | ||
|
|
||
| /** | ||
| * This is currently a no-op on web | ||
| */ | ||
| async refreshSchema(): Promise<void> {} | ||
|
|
||
| async execute(query: string, params?: any[] | undefined): Promise<QueryResult> { | ||
| return this.writeLock((ctx) => ctx.execute(query, params)); | ||
| } | ||
|
|
||
| async executeBatch(query: string, params?: any[][]): Promise<QueryResult> { | ||
| return this.writeLock((ctx) => this._executeBatch(query, params)); | ||
| } | ||
|
|
||
| /** | ||
| * Attempts to close the connection. | ||
| * Shared workers might not actually close the connection if other | ||
| * tabs are still using it. | ||
| */ | ||
| close() { | ||
| this._disposeTableChangeListener?.(); | ||
| this.baseDB?.close?.(); | ||
| } | ||
|
|
||
| async getAll<T>(sql: string, parameters?: any[] | undefined): Promise<T[]> { | ||
| await this.waitForInitialized(); | ||
| return this.dbGetHelpers!.getAll(sql, parameters); | ||
| } | ||
|
|
||
| async getOptional<T>(sql: string, parameters?: any[] | undefined): Promise<T | null> { | ||
| await this.waitForInitialized(); | ||
| return this.dbGetHelpers!.getOptional(sql, parameters); | ||
| } | ||
|
|
||
| async get<T>(sql: string, parameters?: any[] | undefined): Promise<T> { | ||
| await this.waitForInitialized(); | ||
| return this.dbGetHelpers!.get(sql, parameters); | ||
| } | ||
|
|
||
| async readLock<T>(fn: (tx: LockContext) => Promise<T>, options?: DBLockOptions | undefined): Promise<T> { | ||
| await this.waitForInitialized(); | ||
| return this.acquireLock(async () => fn(this.generateDBHelpers({ execute: this._execute }))); | ||
| } | ||
|
|
||
| async writeLock<T>(fn: (tx: LockContext) => Promise<T>, options?: DBLockOptions | undefined): Promise<T> { | ||
| await this.waitForInitialized(); | ||
| return this.acquireLock(async () => fn(this.generateDBHelpers({ execute: this._execute }))); | ||
| } | ||
|
|
||
| protected acquireLock(callback: () => Promise<any>): Promise<any> { | ||
| return getNavigatorLocks().request(`db-lock-${this._dbIdentifier}`, callback); | ||
| } | ||
|
|
||
| async readTransaction<T>(fn: (tx: Transaction) => Promise<T>, options?: DBLockOptions | undefined): Promise<T> { | ||
| return this.readLock(this.wrapTransaction(fn)); | ||
| } | ||
|
|
||
| writeTransaction<T>(fn: (tx: Transaction) => Promise<T>, options?: DBLockOptions | undefined): Promise<T> { | ||
| return this.writeLock(this.wrapTransaction(fn)); | ||
| } | ||
|
|
||
| private generateDBHelpers<T extends { execute: (sql: string, params?: any[]) => Promise<QueryResult> }>( | ||
| tx: T | ||
| ): T & DBGetUtils { | ||
| return { | ||
| ...tx, | ||
| /** | ||
| * Execute a read-only query and return results | ||
| */ | ||
| async getAll<T>(sql: string, parameters?: any[]): Promise<T[]> { | ||
| const res = await tx.execute(sql, parameters); | ||
| return res.rows?._array ?? []; | ||
| }, | ||
|
|
||
| /** | ||
| * Execute a read-only query and return the first result, or null if the ResultSet is empty. | ||
| */ | ||
| async getOptional<T>(sql: string, parameters?: any[]): Promise<T | null> { | ||
| const res = await tx.execute(sql, parameters); | ||
| return res.rows?.item(0) ?? null; | ||
| }, | ||
|
|
||
| /** | ||
| * Execute a read-only query and return the first result, error if the ResultSet is empty. | ||
| */ | ||
| async get<T>(sql: string, parameters?: any[]): Promise<T> { | ||
| const res = await tx.execute(sql, parameters); | ||
| const first = res.rows?.item(0); | ||
| if (!first) { | ||
| throw new Error('Result set is empty'); | ||
| } | ||
| return first; | ||
| } | ||
| }; | ||
| } | ||
|
|
||
| /** | ||
| * Wraps a lock context into a transaction context | ||
| */ | ||
| private wrapTransaction<T>(cb: (tx: Transaction) => Promise<T>) { | ||
| return async (tx: LockContext): Promise<T> => { | ||
| await this._execute('BEGIN TRANSACTION'); | ||
| let finalized = false; | ||
| const commit = async (): Promise<QueryResult> => { | ||
| if (finalized) { | ||
| return { rowsAffected: 0 }; | ||
| } | ||
| finalized = true; | ||
| return this._execute('COMMIT'); | ||
| }; | ||
|
|
||
| const rollback = () => { | ||
| finalized = true; | ||
| return this._execute('ROLLBACK'); | ||
| }; | ||
|
|
||
| try { | ||
| const result = await cb({ | ||
| ...tx, | ||
| commit, | ||
| rollback | ||
| }); | ||
|
|
||
| if (!finalized) { | ||
| await commit(); | ||
| } | ||
| return result; | ||
| } catch (ex) { | ||
| this.logger.debug('Caught ex in transaction', ex); | ||
| try { | ||
| await rollback(); | ||
| } catch (ex2) { | ||
| // In rare cases, a rollback may fail. | ||
| // Safe to ignore. | ||
| } | ||
| throw ex; | ||
| } | ||
| }; | ||
| } | ||
|
|
||
| /** | ||
| * Wraps the worker execute function, awaiting for it to be available | ||
| */ | ||
| private _execute = async (sql: string, bindings?: any[]): Promise<QueryResult> => { | ||
| await this.waitForInitialized(); | ||
| const result = await this.baseDB.execute(sql, bindings); | ||
| return { | ||
| ...result, | ||
| rows: { | ||
| ...result.rows, | ||
| item: (idx: number) => result.rows._array[idx] | ||
| } | ||
| }; | ||
| }; | ||
|
|
||
| /** | ||
| * Wraps the worker executeBatch function, awaiting for it to be available | ||
| */ | ||
| private _executeBatch = async (query: string, params?: any[]): Promise<QueryResult> => { | ||
| await this.waitForInitialized(); | ||
| const result = await this.baseDB.executeBatch(query, params); | ||
| return { | ||
| ...result, | ||
| rows: undefined | ||
| }; | ||
| }; | ||
| } |
21 changes: 21 additions & 0 deletions
21
packages/web/src/db/adapters/ProxiedAsyncDatabaseConnection.ts
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,21 @@ | ||
| import * as Comlink from 'comlink'; | ||
| import { AsyncDatabaseConnection, OnTableChangeCallback } from './AsyncDatabaseConnection'; | ||
|
|
||
| /** | ||
| * @internal | ||
| * Proxies an {@link AsyncDatabaseConnection} which allows for registering table change notification | ||
| * callbacks over a worker channel | ||
| */ | ||
| export function ProxiedAsyncDatabaseConnection(base: AsyncDatabaseConnection) { | ||
| return new Proxy(base, { | ||
| get(target, prop: keyof AsyncDatabaseConnection, receiver) { | ||
| const original = Reflect.get(target, prop, receiver); | ||
| if (typeof original === 'function' && prop === 'registerOnTableChange') { | ||
| return function (callback: OnTableChangeCallback) { | ||
| return base.registerOnTableChange(Comlink.proxy(callback)); | ||
| }; | ||
| } | ||
| return original; | ||
| } | ||
| }); | ||
| } |
Oops, something went wrong.