deno.land / x / replicache@v10.0.0-beta.0 / transactions.ts
نووسراو ببینە
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389import type {LogContext} from '@rocicorp/logger';import {deepClone, JSONValue, ReadonlyJSONValue} from './json';import { isScanIndexOptions, KeyTypeForScanOptions, ScanIndexOptions, ScanOptions, toDbScanOptions,} from './scan-options';import {fromKeyForIndexScanInternal, ScanResultImpl} from './scan-iterator';import type {ScanResult} from './scan-iterator';import {throwIfClosed} from './transaction-closed-error';import * as db from './db/mod';import * as sync from './sync/mod';import type {Hash} from './hash';import type {ScanSubscriptionInfo} from './subscriptions';import type {ScanNoIndexOptions} from './mod.js';import {decodeIndexKey, IndexKey} from './db/index.js';
/** * ReadTransactions are used with [[Replicache.query]] and * [[Replicache.subscribe]] and allows read operations on the * database. */export interface ReadTransaction { readonly clientID: string;
/** * Get a single value from the database. If the `key` is not present this * returns `undefined`. */ get(key: string): Promise<ReadonlyJSONValue | undefined>;
/** Determines if a single `key` is present in the database. */ has(key: string): Promise<boolean>;
/** Whether the database is empty. */ isEmpty(): Promise<boolean>;
/** * Gets many values from the database. This returns a [[ScanResult]] which * implements `AsyncIterable`. It also has methods to iterate over the [[ScanResult.keys|keys]] * and [[ScanResult.entries|entries]]. * * If `options` has an `indexName`, then this does a scan over an index with * that name. A scan over an index uses a tuple for the key consisting of * `[secondary: string, primary: string]`. * * If the [[ScanResult]] is used after the `ReadTransaction` has been closed it * will throw a [[TransactionClosedError]]. */ scan(): ScanResult<string, ReadonlyJSONValue>;
/** * Gets many values from the database. This returns a [[ScanResult]] which * implements `AsyncIterable`. It also has methods to iterate over the [[ScanResult.keys|keys]] * and [[ScanResult.entries|entries]]. * * If `options` has an `indexName`, then this does a scan over an index with * that name. A scan over an index uses a tuple for the key consisting of * `[secondary: string, primary: string]`. * * If the [[ScanResult]] is used after the `ReadTransaction` has been closed it * will throw a [[TransactionClosedError]]. */ scan<Options extends ScanOptions>( options?: Options, ): ScanResult<KeyTypeForScanOptions<Options>, ReadonlyJSONValue>;}
let transactionIDCounter = 0;
export class ReadTransactionImpl< Value extends ReadonlyJSONValue = ReadonlyJSONValue,> implements ReadTransaction{ readonly clientID: string; protected readonly _dbtx: db.Read; protected readonly _lc: LogContext;
constructor( clientID: string, dbRead: db.Read, lc: LogContext, rpcName = 'openReadTransaction', ) { this.clientID = clientID; this._dbtx = dbRead; this._lc = lc .addContext(rpcName) .addContext('txid', transactionIDCounter++); }
async get(key: string): Promise<Value | undefined> { throwIfClosed(this._dbtx); const rv = await this._dbtx.get(key); if (this._dbtx instanceof db.Write) { return (rv && deepClone(rv)) as Value | undefined; } return rv as Value | undefined; }
async has(key: string): Promise<boolean> { throwIfClosed(this._dbtx); return this._dbtx.has(key); }
async isEmpty(): Promise<boolean> { throwIfClosed(this._dbtx); return this._dbtx.isEmpty(); }
scan(): ScanResult<string, Value>; scan<Options extends ScanOptions>( options?: Options, ): ScanResult<KeyTypeForScanOptions<Options>, Value>; scan<Options extends ScanOptions>( options?: Options, ): ScanResult<KeyTypeForScanOptions<Options>, Value> { return scan(options, this._dbtx, noop); }}
function noop(_: unknown): void { // empty}
function scan<Options extends ScanOptions, Value>( options: Options | undefined, dbRead: db.Read, onLimitKey: (inclusiveLimitKey: string) => void,): ScanResult<KeyTypeForScanOptions<Options>, Value> { const iter = getScanIterator(dbRead, options); return makeScanResultFromScanIteratorInternal( iter, options ?? ({} as Options), dbRead, onLimitKey, ) as ScanResult<KeyTypeForScanOptions<Options>, Value>;}
// An implementation of ReadTransaction that keeps track of `keys` and `scans`// for use with Subscriptions.export class SubscriptionTransactionWrapper extends ReadTransactionImpl<ReadonlyJSONValue> { private readonly _keys: Set<string> = new Set(); private readonly _scans: ScanSubscriptionInfo[] = [];
isEmpty(): Promise<boolean> { // Any change to the subscription requires rerunning it. this._scans.push({options: {}}); return super.isEmpty(); }
get(key: string): Promise<ReadonlyJSONValue | undefined> { this._keys.add(key); return super.get(key); }
has(key: string): Promise<boolean> { this._keys.add(key); return super.has(key); }
scan(): ScanResult<string, ReadonlyJSONValue>; scan<Options extends ScanOptions>( options?: Options, ): ScanResult<KeyTypeForScanOptions<Options>, ReadonlyJSONValue>; scan<Options extends ScanOptions>( options?: Options, ): ScanResult<KeyTypeForScanOptions<Options>, ReadonlyJSONValue> { const scanInfo: ScanSubscriptionInfo = { options: toDbScanOptions(options), inclusiveLimitKey: undefined, }; this._scans.push(scanInfo); return scan(options, this._dbtx, inclusiveLimitKey => { scanInfo.inclusiveLimitKey = inclusiveLimitKey; }); }
get keys(): ReadonlySet<string> { return this._keys; }
get scans(): ScanSubscriptionInfo[] { return this._scans; }}
/** * WriteTransactions are used with *mutators* which are registered using * [[ReplicacheOptions.mutators]] and allows read and write operations on the * database. */export interface WriteTransaction extends ReadTransaction { /** * Sets a single `value` in the database. The `value` will be encoded using * `JSON.stringify`. */ put(key: string, value: JSONValue): Promise<void>;
/** * Removes a `key` and its value from the database. Returns `true` if there was a * `key` to remove. */ del(key: string): Promise<boolean>;
/** * Overrides [[ReadTransaction.get]] to return a mutable [[JSONValue]]. */ get(key: string): Promise<JSONValue | undefined>;
/** * Overrides [[ReadTransaction.scan]] to return a mutable [[JSONValue]]. */ scan(): ScanResult<string, JSONValue>; scan<Options extends ScanOptions>( options?: Options, ): ScanResult<KeyTypeForScanOptions<Options>, JSONValue>;}
export class WriteTransactionImpl extends ReadTransactionImpl<JSONValue> implements WriteTransaction{ // use `declare` to specialize the type. protected declare readonly _dbtx: db.Write;
constructor( clientID: string, dbWrite: db.Write, lc: LogContext, rpcName = 'openWriteTransaction', ) { super(clientID, dbWrite, lc, rpcName); }
async put(key: string, value: JSONValue): Promise<void> { throwIfClosed(this._dbtx); await this._dbtx.put(this._lc, key, deepClone(value)); }
async del(key: string): Promise<boolean> { throwIfClosed(this._dbtx); return await this._dbtx.del(this._lc, key); }
async commit( generateChangedKeys: boolean, ): Promise<[Hash, sync.ChangedKeysMap]> { const txn = this._dbtx; throwIfClosed(txn);
const headName = txn.isRebase() ? sync.SYNC_HEAD_NAME : db.DEFAULT_HEAD_NAME; return await txn.commitWithChangedKeys(headName, generateChangedKeys); }}
export interface IndexTransaction extends ReadTransaction { /** * Creates a persistent secondary index in Replicache which can be used with * scan. * * If the named index already exists with the same definition this returns * success immediately. If the named index already exists, but with a * different definition an error is thrown. */ createIndex(def: CreateIndexDefinition): Promise<void>;
/** * Drops an index previously created with [[createIndex]]. */ dropIndex(name: string): Promise<void>;}
/** * The definition of an index. This is used with * [[Replicache.createIndex|createIndex]] when creating indexes. */export interface CreateIndexDefinition { /** The name of the index. This is used when you [[ReadTransaction.scan|scan]] over an index. */ name: string;
/** * The prefix, if any, to limit the index over. If not provided the values of * all keys are indexed. */ prefix?: string;
/** * A [JSON Pointer](https://tools.ietf.org/html/rfc6901) pointing at the sub * value inside each value to index over. * * For example, one might index over users' ages like so: * `createIndex({name: 'usersByAge', keyPrefix: '/user/', jsonPointer: '/age'})` */ jsonPointer: string;}
export class IndexTransactionImpl extends WriteTransactionImpl implements IndexTransaction{ constructor(clientID: string, dbWrite: db.Write, lc: LogContext) { super(clientID, dbWrite, lc, 'openIndexTransaction'); }
async createIndex(options: CreateIndexDefinition): Promise<void> { throwIfClosed(this._dbtx); await this._dbtx.createIndex( this._lc, options.name, options.prefix ?? '', options.jsonPointer, ); }
async dropIndex(name: string): Promise<void> { throwIfClosed(this._dbtx); await this._dbtx.dropIndex(name); }
async commit(): Promise<[Hash, sync.ChangedKeysMap]> { return super.commit(false); }}
type Entry<K> = readonly [key: K, value: ReadonlyJSONValue];
export type IndexKeyEntry = Entry<IndexKey>;
export type StringKeyEntry = Entry<string>;
export type EntryForOptions<Options extends ScanOptions> = Options extends ScanIndexOptions ? IndexKeyEntry : StringKeyEntry;
function getScanIterator<Options extends ScanOptions>( dbRead: db.Read, options: Options | undefined,): AsyncIterable<EntryForOptions<Options>> { if (options && isScanIndexOptions(options)) { return getScanIteratorForIndexMap(dbRead, options) as AsyncIterable< EntryForOptions<Options> >; }
return dbRead.map.scan(fromKeyForNonIndexScan(options)) as AsyncIterable< EntryForOptions<Options> >;}
export function fromKeyForNonIndexScan( options: ScanNoIndexOptions | undefined,): string { if (!options) { return ''; }
const {prefix = '', start} = options; if (start && start.key > prefix) { return start.key; } return prefix;}
function makeScanResultFromScanIteratorInternal< Options extends ScanOptions, Value,>( iter: AsyncIterable<EntryForOptions<Options>>, options: Options, dbRead: db.Read, onLimitKey: (inclusiveLimitKey: string) => void,): ScanResult<KeyTypeForScanOptions<Options>, Value> { return new ScanResultImpl(iter, options, dbRead, onLimitKey);}
async function* getScanIteratorForIndexMap( dbRead: db.Read, options: ScanIndexOptions,): AsyncIterable<IndexKeyEntry> { const map = await dbRead.getMapForIndex(options.indexName); for await (const entry of map.scan(fromKeyForIndexScanInternal(options))) { yield [decodeIndexKey(entry[0]), entry[1]]; }}
replicache
Version Info
2 years ago
