import { Observable } from 'rxjs/Observable' ; import { ErrorObservable } from 'rxjs/observable/ErrorObservable' ; import { Subscription } from 'rxjs/Subscription' ; import { ConnectableObservable } from 'rxjs/observable/ConnectableObservable' ; import * as lf from 'lovefield' ; import * as Exception from '../exception' ; import * as definition from './helper/definition' ; import version from '../version' ; import { Traversable } from '../shared' ; import { Mutation , Selector , QueryToken , PredicateProvider } from './modules' ; import { dispose , contextTableName , fieldIdentifier , hiddenColName } from './symbols' ; import { forEach , clone , contains , tryCatch , hasOwn , getType , assert , identity , warn } from '../utils' ; import { createPredicate , createPkClause , mergeTransactionResult , predicatableQuery , lfFactory } from './helper' ; import { Relationship , RDBType , DataStoreType , LeafType , StatementType , JoinMode } from '../interface/enum' ; import { Record , Fields , JoinInfo , Query , Clause , Predicate } from '../interface' ; import { SchemaDef , ColumnDef , ParsedSchema , Association , ScopedHandler } from '../interface' ; import { ColumnLeaf , NavigatorLeaf , ExecutorResult , UpsertContext , SelectContext } from '../interface' ; export class Database { public static version = version ; public static getTables ( db , ... tableNames ) { return tableNames . map ( ( name ) => db . getSchema ( ) . table ( name ) ) ; } public database$ : ConnectableObservable < lf . Database > ; private schemaDefs = new Map < string , SchemaDef < any > > ( ) ; private schemas = new Map < string , ParsedSchema > ( ) ; private schemaBuilder : lf . schema . Builder | null ; private connected = false ; private storedIds = new Set < string > ( ) ; private subscription : Subscription ; private findPrimaryKey = ( name ) => { return this . findSchema ( name ) ! . pk ; } ; private findSchema = ( name ) : ParsedSchema => { const schema = this . schemas . get ( name ) ; assert ( schema , Exception . NonExistentTable ( name ) ) ; return schema ! ; } ; method description defineSchema < T > ( tableName , schema ) { const advanced = ! this . schemaDefs . has ( tableName ) && ! this . connected ; assert ( advanced , Exception . UnmodifiableTable ( ) ) ; const hasPK = Object . keys ( schema ) . some ( ( key ) => schema [ key ] . primaryKey === true ) ; assert ( hasPK , Exception . PrimaryKeyNotProvided ( ) ) ; this . schemaDefs . set ( tableName , schema ) ; return this ; } constructor param storeType param enableInspector param name param version constructor ( storeType = DataStoreType . MEMORY , enableInspector = false , name = 'ReactiveDB' , version = 1 ) { this . schemaBuilder = lf . schema . create ( name , version ) ; this . database$ = lfFactory ( this . schemaBuilder , { storeType , enableInspector } ) ; } connect ( ) { this . buildTables ( ) ; this . connected = true ; this . schemaDefs . clear ( ) ; this . subscription = this . database$ . connect ( ) ; } dump ( ) { return this . database$ . concatMap ( db => db . export ( ) ) ; } load ( data ) { assert ( ! this . connected , Exception . DatabaseIsNotEmpty ( ) ) ; return this . database$ . concatMap ( db => { forEach ( data . tables , ( entities , name ) => { const schema = this . findSchema ( name ) ; entities . forEach ( ( entity ) => this . storedIds . add ( fieldIdentifier ( name , entity [ schema . pk ] ) ) ) ; } ) ; return db . import ( data ) . catch ( ( ) => { forEach ( data . tables , ( entities , name ) => { const schema = this . findSchema ( name ) ; entities . forEach ( ( entity ) => this . storedIds . delete ( fieldIdentifier ( name , entity [ schema . pk ] ) ) ) ; } ) ; } ) ; } ) ; } insert < T > ( tableName , raw ) ; insert < T > ( tableName , raw ) ; insert < T > ( tableName , raw ) ; insert < T > ( tableName , raw ) { return this . database$ . concatMap ( db => { const schema = this . findSchema ( tableName ) ; const pk = schema . pk ; const columnMapper = schema . mapper ; const [ table ] = Database . getTables ( db , tableName ) ; const muts = [ ] ; const entities = clone ( raw ) ; const iterator = Array . isArray ( entities ) ? entities : [ entities ] ; iterator . forEach ( ( entity ) => { const mut = new Mutation ( db , table ) ; const hiddenPayload = Object . create ( null ) ; columnMapper . forEach ( ( mapper , key ) => { if ( ! hasOwn ( entity , key ) || key === pk ) { return ; } const val = entity [ key ] ; hiddenPayload [ key ] = mapper ( val ) ; hiddenPayload [ hiddenColName ( key ) ] = val ; } ) ; mut . patch ( { ... entity , ... hiddenPayload } ) ; mut . withId ( pk , entity [ pk ] ) ; muts . push ( mut ) ; } ) ; const { contextIds , queries } = Mutation . aggregate ( db , muts , [ ] ) ; contextIds . forEach ( id => this . storedIds . add ( id ) ) ; return this . executor ( db , queries ) . do ( { error : ( ) => contextIds . forEach ( id => this . storedIds . delete ( id ) ) } ) ; } ) ; } get < T > ( tableName , query = { } , mode = JoinMode . imlicit ) { const selector$ = this . buildSelector < T > ( tableName , query , mode ) ; return new QueryToken < T > ( selector$ ) ; } update < T > ( tableName , clause , raw ) { const type = getType ( raw ) ; if ( type !== 'Object' ) { return Observable . throw ( Exception . InvalidType ( [ 'Object' , type ] ) ) ; } const [ schema , err ] = tryCatch < ParsedSchema > ( this . findSchema ) ( tableName ) ; if ( err ) { return Observable . throw ( err ) ; } return this . database$ . concatMap < any , any > ( db => { const entity = clone ( raw ) ; const [ table ] = Database . getTables ( db , tableName ) ; const columnMapper = schema ! . mapper ; const hiddenPayload = Object . create ( null ) ; columnMapper . forEach ( ( mapper , key ) => { if ( ! hasOwn ( entity , key ) || key === schema ! . pk ) { return ; } const val = ( entity as any ) [ key ] ; hiddenPayload [ key ] = mapper ( val ) ; hiddenPayload [ hiddenColName ( key ) ] = val ; } ) ; const mut = { ... ( entity as any ) , ... hiddenPayload } ; const predicate = createPredicate ( table , clause ) ; const query = predicatableQuery ( db , table , predicate ! , StatementType . Update ) ; forEach ( mut , ( val , key ) => { const column = table [ key ] ; if ( key === schema ! . pk ) { warn ( `Primary key is not modifiable.` ) ; } else if ( ! column ) { warn ( `Column: ${ key } is not existent on table:${ tableName }` ) ; } else { query . set ( column , val ) ; } } ) ; return this . executor ( db , [ query ] ) ; } ) ; } delete < T > ( tableName , clause = { } ) { const [ pk , err ] = tryCatch < string > ( this . findPrimaryKey ) ( tableName ) ; if ( err ) { return Observable . throw ( err ) ; } return this . database$ . concatMap ( db => { const [ table ] = Database . getTables ( db , tableName ) ; const column = table [ pk ! ] ; const provider = new PredicateProvider ( table , clause ) ; const prefetch = predicatableQuery ( db , table , provider . getPredicate ( ) , StatementType . Select , column ) ; return Observable . fromPromise ( prefetch . exec ( ) ) . concatMap ( ( scopedIds ) => { const query = predicatableQuery ( db , table , provider . getPredicate ( ) , StatementType . Delete ) ; scopedIds . forEach ( ( entity ) => this . storedIds . delete ( fieldIdentifier ( tableName , entity [ pk ! ] ) ) ) ; return this . executor ( db , [ query ] ) . do ( { error : ( ) => { scopedIds . forEach ( ( entity ) => this . storedIds . add ( fieldIdentifier ( tableName , entity [ pk ! ] ) ) ) ; } } ) ; } ) ; } ) ; } upsert < T > ( tableName , raw ) ; upsert < T > ( tableName , raw ) ; upsert < T > ( tableName , raw ) ; upsert < T > ( tableName , raw ) { return this . database$ . concatMap ( db => { const sharing = new Map < any , Mutation > ( ) ; const insert = [ ] ; const update = [ ] ; this . traverseCompound ( db , tableName , clone ( raw ) , insert , update , sharing ) ; const { contextIds , queries } = Mutation . aggregate ( db , insert , update ) ; if ( queries . length > 0 ) { contextIds . forEach ( id => this . storedIds . add ( id ) ) ; return this . executor ( db , queries ) . do ( { error : ( ) => contextIds . forEach ( id => this . storedIds . delete ( id ) ) } ) ; } else { return Observable . of ( { result : false , insert : 0 , update : 0 , delete : 0 , select : 0 } ) ; } } ) ; } remove < T > ( tableName , clause = { } ) { const [ schema , err ] = tryCatch < ParsedSchema > ( this . findSchema ) ( tableName ) ; if ( err ) { return Observable . throw ( err ) ; } const disposeHandler = schema ! . dispose ; return this . database$ . concatMap ( ( db ) => { const [ table ] = Database . getTables ( db , tableName ) ; const predicate = createPredicate ( table , clause . where ) ; const queries = [ ] ; const removedIds = [ ] ; queries . push ( predicatableQuery ( db , table , predicate ! , StatementType . Delete ) ) ; const prefetch = predicatableQuery ( db , table , predicate ! , StatementType . Select ) ; return Observable . fromPromise ( prefetch . exec ( ) ) . concatMap ( ( rootEntities ) => { rootEntities . forEach ( entity => { removedIds . push ( fieldIdentifier ( tableName , entity [ schema ! . pk ] ) ) ; } ) ; if ( disposeHandler ) { const scope = this . createScopedHandler < T > ( db , queries , removedIds ) ; return disposeHandler ( rootEntities , scope ) . do ( ( ) => removedIds . forEach ( ( id ) => this . storedIds . delete ( id ) ) ) . concatMap ( ( ) => this . executor ( db , queries ) ) ; } else { removedIds . forEach ( ( id ) => this . storedIds . delete ( id ) ) ; return this . executor ( db , queries ) ; } } ) . do ( { error : ( ) => removedIds . forEach ( ( id ) => this . storedIds . add ( id ) ) } ) ; } ) ; } dispose ( ) { if ( ! this . connected ) { return Observable . throw ( Exception . NotConnected ( ) ) ; } const cleanup = this . database$ . map ( db => db . getSchema ( ) . tables ( ) . map ( t => db . delete ( ) . from ( t ) ) ) ; return this . database$ . concatMap ( db => { return cleanup . concatMap ( queries => this . executor ( db , queries ) ) . do ( ( ) => { db . close ( ) ; this . schemas . clear ( ) ; this . storedIds . clear ( ) ; this . schemaBuilder = null ; this . subscription . unsubscribe ( ) ; } ) ; } ) ; } private buildTables ( ) { this . schemaDefs . forEach ( ( schemaDef , tableName ) => { const tableBuilder = this . schemaBuilder ! . createTable ( tableName ) ; this . parseSchemaDef ( tableName , schemaDef , tableBuilder ) ; } ) ; } /**
     * 解析 schemaDefs, 根据解析后的 metadata 建表
     */ private parseSchemaDef ( tableName , schemaDef , tableBuilder ) { const uniques = [ ] ; const indexes = [ ] ; const primaryKey = [ ] ; const nullable = [ ] ; const columns = new Map < string , RDBType > ( ) ; const associations = new Map < string , Association > ( ) ; const mapper = new Map < string , Function > ( ) ; const disposeHandler = ( typeof schemaDef . dispose === 'function' && schemaDef . dispose ) || ( typeof schemaDef [ dispose ] === 'function' && schemaDef [ dispose ] ) || null ; forEach ( schemaDef , ( def , key ) => { if ( typeof def === 'function' ) { return ; } if ( ! def . virtual ) { this . createColumn ( tableBuilder , key , def . type as RDBType , nullable , mapper ) ; columns . set ( key , def . type ) ; if ( def . primaryKey ) { assert ( ! primaryKey [ 0 ] , Exception . PrimaryKeyConflict ( ) ) ; primaryKey . push ( key ) ; } if ( def . unique ) { uniques . push ( key ) ; } if ( def . index ) { indexes . push ( key ) ; } const isNullable = ! [ def . primaryKey , def . index , def . unique ] . some ( identity ) ; if ( isNullable ) { nullable . push ( key ) ; } } else { associations . set ( key , { where : def . virtual . where , type : def . type as Relationship , name : def . virtual . name } ) ; } } ) ; this . schemas . set ( tableName , { pk : primaryKey [ 0 ] , mapper , columns , dispose : disposeHandler , associations } ) ; if ( indexes . length ) { tableBuilder . addIndex ( 'index' , indexes ) ; } if ( uniques . length ) { tableBuilder . addUnique ( 'unique' , uniques ) ; } if ( nullable . length ) { tableBuilder . addNullable ( nullable ) ; } tableBuilder . addPrimaryKey ( primaryKey ) ; } private buildSelector < T > ( tableName , clause , mode ) { return this . database$ . map ( ( db ) => { const schema = this . findSchema ( tableName ) ; const pk = schema . pk ; const containFields = ! ! clause . fields ; const containKey = containFields ? contains ( pk , clause . fields ! ) : true ; const fields = containFields ? new Set ( clause . fields ) : new Set ( schema . columns . keys ( ) ) ; const { table , columns , joinInfo , definition } = this . traverseQueryFields ( db , tableName , fields , containKey , ! containFields , [ ] , { } , mode ) ; const query = predicatableQuery ( db , table ! , null , StatementType . Select , ... columns ) ; joinInfo . forEach ( ( info ) => query . leftOuterJoin ( info . table , info . predicate ) ) ; const orderDesc = ( clause . orderBy || [ ] ) . map ( desc => { return { column : table ! [ desc . fieldName ] , orderBy : ! desc . orderBy ? null : lf . Order [ desc . orderBy ] } ; } ) ; const matcher = { pk : { name : pk , queried : containKey } , definition , mainTable : table ! } ; const { limit , skip } = clause ; const provider = new PredicateProvider ( table ! , clause . where ! ) ; return new Selector < T > ( db , query , matcher , provider , limit , skip , orderDesc ) ; } ) ; } private createColumn ( tableBuilder , columnName , rdbType , nullable , mapper ) { const hiddenName = hiddenColName ( columnName ) ; switch ( rdbType ) { case RDBType . ARRAY_BUFFER : return tableBuilder . addColumn ( columnName , lf . Type . ARRAY_BUFFER ) ; case RDBType . BOOLEAN : return tableBuilder . addColumn ( columnName , lf . Type . BOOLEAN ) ; case RDBType . DATE_TIME : nullable . push ( hiddenName ) ; mapper . set ( columnName , ( val ) => val ? new Date ( val ) . valueOf ( ) : new Date ( 0 ) . valueOf ( ) ) ; return tableBuilder . addColumn ( columnName , lf . Type . INTEGER ) . addColumn ( hiddenName , lf . Type . STRING ) ; case RDBType . INTEGER : return tableBuilder . addColumn ( columnName , lf . Type . INTEGER ) ; case RDBType . LITERAL_ARRAY : nullable . push ( hiddenName ) ; mapper . set ( columnName , ( val ) => val ? val . join ( '|' ) : '' ) ; return tableBuilder . addColumn ( columnName , lf . Type . STRING ) . addColumn ( hiddenName , lf . Type . OBJECT ) ; case RDBType . NUMBER : return tableBuilder . addColumn ( columnName , lf . Type . NUMBER ) ; case RDBType . OBJECT : return tableBuilder . addColumn ( columnName , lf . Type . OBJECT ) ; case RDBType . STRING : return tableBuilder . addColumn ( columnName , lf . Type . STRING ) ; default : throw Exception . InvalidType ( ) ; } } private traverseQueryFields ( db , tableName , fieldsValue , hasKey , glob , path = [ ] , context = { } , mode ) { const schema = this . findSchema ( tableName ) ; const rootDefinition = Object . create ( null ) ; const navigators = [ ] ; const columns = [ ] ; const joinInfo = [ ] ; if ( mode === JoinMode . imlicit && contains ( tableName , path ) ) { return { columns , joinInfo , advanced : false , table : null , definition : null } ; } else { path . push ( tableName ) ; } schema . associations . forEach ( ( _ , nav ) => { if ( glob && ! contains ( nav , path ) ) { fieldsValue . add ( nav ) ; } navigators . push ( nav ) ; } ) ; const onlyNavigator = Array . from ( fieldsValue . keys ( ) ) . every ( key => contains ( key , navigators ) ) ; assert ( ! onlyNavigator , Exception . InvalidQuery ( ) ) ; if ( ! hasKey ) { fieldsValue . add ( schema . pk ) ; } const suffix = ( context [ tableName ] || 0 ) + 1 ; context [ tableName ] = suffix ; const contextName = contextTableName ( tableName , suffix ) ; const currentTable = Database . getTables ( db , tableName ) [ 0 ] . as ( contextName ) ; const handleAdvanced = ( ret , key , defs ) => { if ( ! ret . advanced ) { return ; } columns . push ( ... ret . columns ) ; assert ( ! rootDefinition [ key ] , Exception . AliasConflict ( key , tableName ) ) ; if ( ( defs as ColumnDef ) . column ) { rootDefinition [ key ] = defs ; } else { const { where , type } = defs as Association ; rootDefinition [ key ] = definition . revise ( type ! , ret . definition ) ; const [ predicate , err ] = tryCatch ( createPredicate ) ( currentTable , where ( ret . table ) ) ; if ( err ) { warn ( `Failed to build predicate, since ${ err . message }` + `, on table: ${ ret . table . getName ( ) }` ) ; } const joinLink = predicate ? [ { table : ret . table , predicate } , ... ret . joinInfo ] : ret . joinInfo ; joinInfo . push ( ... joinLink ) ; } } ; const traversable = new Traversable < SelectContext > ( fieldsValue ) ; traversable . context ( ( field , val , ctx ) => { if ( ! ctx || ctx . isRoot || typeof field !== 'string' ) { return false ; } const isNavigatorLeaf = contains ( field , navigators ) ; const type = isNavigatorLeaf ? LeafType . navigator : LeafType . column ; const key = ctx . key ? ctx . key : val ; if ( isNavigatorLeaf ) { const description = schema . associations . get ( ctx . key ) ; if ( ! description ) { warn ( `Build a relationship failed, field: ${ ctx . key }.` ) ; return false ; } return { type , key , leaf : this . navigatorLeaf ( description , ctx . key , val ) } ; } if ( ! currentTable [ field ] ) { warn ( `Column: ${ field } is not exist on ${ tableName }` ) ; return false ; } return { type , key , leaf : this . columnLeaf ( currentTable , contextName , field ) } ; } ) ; traversable . forEach ( ( ctx ) => { switch ( ctx . type ) { case LeafType . column : const { column , identifier } = ctx . leaf as ColumnLeaf ; const type = schema . columns . get ( ctx . key ) ! ; const columnDef = definition . create ( identifier , schema . pk === ctx . key , type ) ; handleAdvanced ( { columns : [ column ] , advanced : true } , ctx . key , columnDef ) ; break ; case LeafType . navigator : const { containKey , fields , assocaiation } = ctx . leaf as NavigatorLeaf ; const ret = this . traverseQueryFields ( db , assocaiation . name , new Set ( fields ) , containKey , glob , path . slice ( 0 ) , context , mode ) ; handleAdvanced ( ret , ctx . key , assocaiation ) ; ctx . skip ( ) ; break ; } } ) ; return { columns , joinInfo , advanced : true , table : currentTable , definition : rootDefinition } ; } private traverseCompound ( db , tableName , compoundEntites , insertMutList , updateMutList , sharing ) { if ( compoundEntites == null ) { return ; } if ( Array . isArray ( compoundEntites ) ) { compoundEntites . forEach ( ( item ) => this . traverseCompound ( db , tableName , item , insertMutList , updateMutList , sharing ) ) ; return ; } const schema = this . findSchema ( tableName ) ; const pk = schema . pk ; const pkVal = compoundEntites [ pk ] ; assert ( pkVal !== undefined , Exception . PrimaryKeyNotProvided ( ) ) ; const [ table ] = Database . getTables ( db , tableName ) ; const identifier = fieldIdentifier ( tableName , pkVal ) ; const visited = contains ( identifier , sharing ) ; const stored = contains ( identifier , this . storedIds ) ; const mut = visited ? sharing . get ( identifier ) ! : new Mutation ( db , table ) ; if ( ! visited ) { const list = stored ? updateMutList : insertMutList ; list . push ( mut ) ; sharing . set ( identifier , mut ) ; } const traversable = new Traversable < UpsertContext > ( compoundEntites ) ; traversable . context ( ( key , _ , ctx ) => { const isNavigator = schema . associations . has ( key ) ; const isColumn = schema . columns . has ( key ) ; const mapper = ( isColumn && schema . mapper . get ( key ) ) || null ; if ( ! ( isColumn || isNavigator || ctx ! . isRoot ) ) { ctx ! . skip ( ) ; } return ( ctx ! . isRoot || ( ! isColumn && ! isNavigator ) ) ? false : { mapper , visited , isNavigatorLeaf : isNavigator } ; } ) ; traversable . forEach ( ( ctx , node ) => { ctx . skip ( ) ; if ( ctx . isNavigatorLeaf ) { const ref = schema . associations . get ( ctx . key ) ! . name ; return this . traverseCompound ( db , ref , node , insertMutList , updateMutList , sharing ) ; } if ( ctx . key !== pk ) { const res = ctx . mapper ? { [ ctx . key ] : ctx . mapper ( node ) , [ hiddenColName ( ctx . key ) ] : node } : { [ ctx . key ] : node } ; mut . patch ( res ) ; } else if ( ctx . key === pk && ! ctx . visited ) { mut . withId ( ctx . key , node ) ; } } ) ; } private columnLeaf ( table , tableName , key ) { const column = table [ key ] ; const hiddenName = hiddenColName ( key ) ; const identifier = fieldIdentifier ( tableName , key ) ; const hiddenCol = table [ hiddenName ] ; const ret = hiddenCol ? hiddenCol . as ( identifier ) : column . as ( identifier ) ; return { identifier , column : ret } ; } private navigatorLeaf ( assocaiation , _ , val ) { const schema = this . findSchema ( assocaiation . name ) ; const fields = typeof val === 'string' ? new Set ( schema . columns . keys ( ) ) : val ; return { fields , assocaiation , containKey : contains ( schema . pk , val ) } ; } private createScopedHandler < T > ( db , queryCollection , keys ) { return ( tableName ) : ScopedHandler => { const pk = this . findPrimaryKey ( tableName ) ; const remove = ( entities ) => { const [ table ] = Database . getTables ( db , tableName ) ; entities . forEach ( entity => { const pkVal = entity [ pk ] ; const clause = createPkClause ( pk , pkVal ) ; const predicate = createPredicate ( table , clause ) ; const query = predicatableQuery ( db , table , predicate ! , StatementType . Delete ) ; queryCollection . push ( query ) ; keys . push ( fieldIdentifier ( tableName , pkVal ) ) ; } ) ; } ; const get = ( where = null ) => { const [ table ] = Database . getTables ( db , tableName ) ; const [ predicate , err ] = tryCatch ( createPredicate ) ( table , where ) ; if ( err ) { return Observable . throw ( err ) ; } const query = predicatableQuery ( db , table , predicate ! , StatementType . Select ) ; return Observable . fromPromise < T [ ] > ( query . exec ( ) as any ) ; } ; return [ get , remove ] ; } ; } private executor ( db , queries ) { const tx = db . createTransaction ( ) ; const handler = { error : ( ) => warn ( `Execute failed, transaction is already marked for rollback.` ) } ; return Observable . fromPromise ( tx . exec ( queries ) ) . do ( handler ) . map ( ( ret ) => { return { result : true , ... mergeTransactionResult ( queries , ret ) } ; } ) ; } }