import { grpc } from '@improbable-eng/grpc-web' ; import { Context } from '@textile/context' ; import type { Identity } from '@textile/crypto' ; import { KeyInfo , UserAuth } from '@textile/security' ; import { DBInfo , WriteTransaction } from '@textile/threads-client' ; import ThreadID from '@textile/threads-id' ; import type { Dexie } from 'dexie' ; import jsonpatch from 'fast-json-patch' ; import { Change , ChangeTableName , MetaTableName , StashTableName , } from '../middleware/changes' ; import { CollectionConfig , createDbClient , defaults , getToken , getTokenChallenge , GrpcConfig , newDB , } from './grpc' ; export const Errors = { ThreadIDError : new Error ( 'Missing/invalid thread id' ) , MissingDbError : new Error ( 'Missing remote db/thread' ) , NoRemoteError : new Error ( 'No remote service host specified' ) , RemoteError : new Error ( 'Remote operation failed' ) , ChangeError : new Error ( 'Unable to process db change(s)' ) , LocalChangesError : new Error ( 'Unpushed local changes' ) , ThreadExists : new Error ( 'Remote db/thread already exists' ) , } ; export const ThreadIDName = 'thread-id' ; const encoder = new TextEncoder ( ) ; const decoder = new TextDecoder ( ) ; /**
 * Config specifies the configuration options for remote sync.
 */ export interface RemoteConfig extends GrpcConfig { /**
     * Thread id
     */ id : string ; /**
     * Authorization token. This is here for ease of use.
     */ token : string ; } export class Remote { /**
     * Database identity on remote peer. String encoding of a thread id.
     */ public id ? : string ; /**
     * Set of configuration options for remote sync.
     */ public config : Partial < GrpcConfig > ; param storage param config constructor ( private storage , config = defaults ) { const { id , token , ... rest } = config ; this . config = { metadata : new grpc . Metadata ( ) , ... rest } ; if ( token ) { this . config . metadata ?. set ( 'authorization' , `bearer ${ token }` ) ; } this . id = id ; } param auth example example async setUserAuth ( auth ) { const init = typeof auth === 'object' ? Context . fromUserAuth ( auth , this . config . serviceHost ) : Context . fromUserAuthCallback ( auth , this . config . serviceHost ) ; const json = { } ; this . config . metadata ?. forEach ( ( key , values ) => ( json [ key ] = values ) ) ; const metadata = await Context . fromJSON ( { ... json , ... init . toJSON ( ) , } ) . toMetadata ( ) ; this . config . metadata = metadata ; return this ; } param key param host param debug example async setKeyInfo ( key ) { const init = await new Context ( this . config . serviceHost ) . withKeyInfo ( key ) ; const json = { } ; this . config . metadata ?. forEach ( ( key , values ) => ( json [ key ] = values ) ) ; const metadata = await Context . fromJSON ( { ... json , ... init . toJSON ( ) , } ) . toMetadata ( ) ; this . config . metadata = metadata ; return this ; return this ; } param config example set ( config ) { const { id , ... rest } = config ; this . config = { ... this . config , ... rest } ; this . id = id ?? this . id ; return this ; } example get ( ) { const { id , config } = this ; return { id , ... config } ; } async info ( ) { if ( this . id === undefined ) throw Errors . ThreadIDError ; const threadID = ThreadID . fromString ( this . id ) ; const client = createDbClient ( this . config ) ; return client . getDBInfo ( threadID ) ; } param identity param callback see getToken or  getTokenChallenge for lower-level access.
     *  note example async authorize ( identity ) ; async authorize ( identity , callback ) ; async authorize ( identity , callback ? ) { if ( ! this . config . serviceHost ) { throw Errors . NoRemoteError ; } let token = '' ; if ( typeof identity === 'string' ) { if ( callback === undefined ) { throw new Error ( 'Callback required for public key challenge' ) ; } token = await getTokenChallenge ( identity , callback , this . config ) ; } else { token = await getToken ( identity , this . config ) ; } this . config . metadata ?. set ( 'authorization' , `bearer ${ token }` ) ; return token ; } note param name param id async initialize ( id = this . id ) { if ( ! this . config . serviceHost ) { throw Errors . NoRemoteError ; } const meta = this . storage . table ( MetaTableName ) ; let threadID = ThreadID . fromRandom ( ) ; if ( id !== undefined ) { if ( typeof id === 'string' ) { threadID = ThreadID . fromString ( id ) ; } else { threadID = id ; } } else { const { value } = ( await meta . get ( ThreadIDName ) ) ?? { } ; if ( value !== undefined ) { threadID = ThreadID . fromString ( value ) ; } } const schemas = await Promise . all ( this . storage . tables . filter ( ( table ) => ! table . name . startsWith ( '_' ) ) . map ( async ( table ) => ( { name : table . name , schema : encoder . encode ( JSON . stringify ( await table . getSchema ( ) ) ) , indexesList : [ ] , writevalidator : '' , readfilter : '' , } ) ) ) ; let idString = '' ; try { idString = await newDB ( this . storage . name , threadID , schemas , this . config ) ; } catch ( err ) { const msg = err . toString ( ) ; if ( msg . includes ( 'db already exists' ) || msg . includes ( 'multiple write errors' ) ) { idString = threadID . toString ( ) ; const client = createDbClient ( this . config ) ; for ( const schema of schemas ) { schema . schema = JSON . parse ( decoder . decode ( schema . schema ) ) ; try { await client . newCollection ( threadID , schema ) ; } catch ( err ) { if ( ! err . message . includes ( 'collection already registered' ) ) throw err ; } } } else { throw err ; } } if ( ! idString ) throw Errors . RemoteError ; this . set ( { id : idString } ) ; await meta . put ( { key : ThreadIDName , value : idString } ) ; return idString ; } async push ( ... collections ) { if ( ! this . config . serviceHost ) { throw Errors . NoRemoteError ; } if ( this . id === undefined ) throw Errors . ThreadIDError ; const threadID = ThreadID . fromString ( this . id ) ; const localChanges = this . storage . table < Change > ( ChangeTableName ) ; const client = createDbClient ( this . config ) ; for ( const collectionName of collections ) { this . storage . table ( collectionName ) ; const filtered = localChanges . where ( 'name' ) . equals ( collectionName ) ; if ( ( await filtered . count ( ) ) < 1 ) { return ; } let trans ; try { const changes = await filtered . toArray ( ) ; let count = 0 ; for ( const obj of changes ) { trans = client . writeTransaction ( threadID , collectionName ) ; await trans . start ( ) ; switch ( obj . type ) { case 'put' : { try { await trans . save ( [ obj . after ] ) ; break ; } catch ( err ) { throw err ; } } case 'add' : { try { await trans . create ( [ obj . after ] ) ; break ; } catch ( err ) { throw err ; } } case 'delete' : { try { await trans . delete ( [ obj . key ] ) ; break ; } catch ( err ) { throw err ; } } } await trans . end ( ) ; count ++ ; } const deleted = await filtered . delete ( ) ; if ( count !== deleted ) throw Errors . ChangeError ; } catch ( err ) { try { await trans ?. discard ( ) ; } catch ( err ) { } throw err ; } } } see Remote . applyStash .
      async createStash ( ) { const changes = this . storage . table ( ChangeTableName ) ; if ( ! ( await changes . count ( ) ) ) return ; const stash = this . storage . table ( StashTableName ) ; const array = await changes . toArray ( ) ; await stash . bulkPut ( array ) ; return changes . clear ( ) ; } see Remote . createStash .
      async clearStash ( ) { const stash = this . storage . table ( StashTableName ) ; return stash . clear ( ) ; } param collections see Remote . createStash .
      async applyStash ( ... collections ) { for ( const collectionName of collections ) { await this . storage . transaction ( 'rw' , [ collectionName , StashTableName ] , async ( tx ) => { const stash = tx . table < Change , string > ( StashTableName ) ; const table = tx . table ( collectionName ) ; const filtered = stash . where ( 'name' ) . equals ( collectionName ) ; if ( ( await filtered . count ( ) ) < 1 ) { return ; } for ( const obj of await filtered . toArray ( ) ) { if ( obj . type === 'delete' ) { table . delete ( obj . key ) ; } else { const value = ( await table . get ( obj . key ) ) ?? { } ; jsonpatch . applyPatch ( value , obj . ops , false , true ) ; await table . put ( value ) ; } } await filtered . delete ( ) ; } ) ; } } note param collections returns async pull ( ... collections ) { if ( ! this . config . serviceHost ) { throw Errors . NoRemoteError ; } if ( this . id === undefined ) throw Errors . ThreadIDError ; const threadID = ThreadID . fromString ( this . id ) ; const localChanges = this . storage . table < Change > ( ChangeTableName ) ; if ( await localChanges . count ( ) ) throw Errors . LocalChangesError ; const client = createDbClient ( this . config ) ; for ( const collectionName of collections ) { const instances = await client . find ( threadID , collectionName , { } ) ; const table = this . storage . table ( collectionName ) ; const keys = await table . bulkPut ( instances , { allKeys : true } ) ; await table . filter ( ( obj ) => ! keys . includes ( obj . _id ) ) . delete ( ) ; } const isModUpdate = ( ops ) : boolean => { const [ op0 ] = ops ; return op0 . op === 'add' && op0 . path == '/_mod' ; } ; const changes = this . storage . table < Change , string > ( ChangeTableName ) ; const values = await changes . filter ( ( change ) => change . ops . length > 0 && ! isModUpdate ( change . ops ) ) . toArray ( ) ; await changes . clear ( ) ; return values . map ( ( change ) => change . key ) ; } }