import { DynamoDB } from 'aws-sdk' ; import { EntityAttributes , EntityTarget , MANAGER_NAME , QUERY_ORDER , STATS_TYPE , } from '@typedorm/common' ; import { getDynamoQueryItemsLimit } from '../../helpers/get-dynamo-query-items-limit' ; import { isEmptyObject } from '../../helpers/is-empty-object' ; import { Connection } from '../connection/connection' ; import { DocumentClientRequestTransformer } from '../transformer/document-client-request-transformer' ; import { EntityTransformer } from '../transformer/entity-transformer' ; import { getConstructorForInstance } from '../../helpers/get-constructor-for-instance' ; import { isUsedForPrimaryKey } from '../../helpers/is-used-for-primary-key' ; import { isWriteTransactionItemList } from '../transaction/type-guards' ; import { isLazyTransactionWriteItemListLoader } from '../transformer/is-lazy-transaction-write-item-list-loader' ; import { FilterOptions } from '../expression/filter-options-type' ; import { ConditionOptions } from '../expression/condition-options-type' ; import { MetadataOptions } from '../transformer/base-transformer' ; import { getUniqueRequestId } from '../../helpers/get-unique-request-id' ; import { ProjectionKeys } from '../expression/projection-keys-options-type' ; import { KeyConditionOptions } from '../expression/key-condition-options-type' ; import { UpdateBody } from '../expression/update-body-type' ; export interface EntityManagerCreateOptions < Entity > { default overwriteIfExists ? : boolean ; /**
     * Specify condition to apply
     */ where ? : ConditionOptions < Entity > ; } export interface EntityManagerUpdateOptions < Entity > { default nestedKeySeparator ? : string ; /**
     * Specify condition to apply
     */ where ? : ConditionOptions < Entity > ; } export interface EntityManagerDeleteOptions < Entity > { /**
     * Specify condition to apply
     */ where ? : ConditionOptions < Entity > ; } export interface EntityManagerFindOptions < Entity , PartitionKey > { /**
     * Index to query, when omitted, query will be run against main table
     */ queryIndex ? : string ; default keyCondition ? : KeyConditionOptions ; default limit ? : number ; default orderBy ? : QUERY_ORDER ; /**
     * Cursor to traverse from
     */ cursor ? : DynamoDB . DocumentClient . Key ; /**
     * Specify filter to apply
     * Avoid using this where possible, since filters in dynamodb applies after items
     * are read
     */ where ? : FilterOptions < Entity , PartitionKey > ; default select ? : ProjectionKeys < Entity > ; } export interface EntityManagerCountOptions < Entity , PartitionKey > { /**
     * Index to query, when omitted, query will be run against main table
     */ queryIndex ? : string ; default keyCondition ? : KeyConditionOptions ; /**
     * Specify filter to apply
     * Avoid using this where possible, since filters in dynamodb applies after items
     * are read
     */ where ? : FilterOptions < Entity , PartitionKey > ; } export interface EntityManagerFindOneOptions < Entity > { default select ? : ProjectionKeys < Entity > ; } export class EntityManager { private _dcReqTransformer : DocumentClientRequestTransformer ; private _entityTransformer : EntityTransformer ; constructor ( protected connection ) { this . _dcReqTransformer = new DocumentClientRequestTransformer ( connection ) ; this . _entityTransformer = new EntityTransformer ( connection ) ; } param entity async create < Entity > ( entity , options ? , metadataOptions ? ) { const requestId = getUniqueRequestId ( metadataOptions ?. requestId ) ; const dynamoPutItemInput = this . _dcReqTransformer . toDynamoPutItem ( entity , options , { requestId , returnConsumedCapacity : metadataOptions ?. returnConsumedCapacity , } ) ; const entityClass = getConstructorForInstance ( entity ) ; if ( ! isWriteTransactionItemList ( dynamoPutItemInput ) ) { const response = await this . connection . documentClient . put ( dynamoPutItemInput ) . promise ( ) ; if ( response ?. ConsumedCapacity ) { this . connection . logger . logStats ( { requestId , scope : MANAGER_NAME . ENTITY_MANAGER , statsType : STATS_TYPE . CONSUMED_CAPACITY , consumedCapacityData : response . ConsumedCapacity , } ) ; } const itemToReturn = this . _entityTransformer . fromDynamoEntity < Entity > ( entityClass , dynamoPutItemInput . Item , { requestId , } ) ; return itemToReturn ; } await this . connection . transactionManger . writeRaw ( dynamoPutItemInput , { requestId , returnConsumedCapacity : metadataOptions ?. returnConsumedCapacity , } ) ; const itemToReturn = this . _entityTransformer . fromDynamoEntity < Entity > ( entityClass , dynamoPutItemInput [ 0 ] ?. Put ?. Item ?? { } , { requestId , } ) ; return itemToReturn ; } param entityClass param props async findOne < Entity , PrimaryKey = Partial < Entity > > ( entityClass , primaryKeyAttributes , options ? , metadataOptions ? ) { const requestId = getUniqueRequestId ( metadataOptions ?. requestId ) ; const dynamoGetItem = this . _dcReqTransformer . toDynamoGetItem ( entityClass , primaryKeyAttributes , options , { requestId , returnConsumedCapacity : metadataOptions ?. returnConsumedCapacity , } ) ; const response = await this . connection . documentClient . get ( dynamoGetItem ) . promise ( ) ; if ( response ?. ConsumedCapacity ) { this . connection . logger . logStats ( { requestId , scope : MANAGER_NAME . ENTITY_MANAGER , statsType : STATS_TYPE . CONSUMED_CAPACITY , consumedCapacityData : response . ConsumedCapacity , } ) ; } if ( ! response . Item ) { return ; } const entity = this . _entityTransformer . fromDynamoEntity < Entity > ( entityClass , response . Item , { requestId , } ) ; return entity ; } param entityClass param attributes async exists < Entity , KeyAttributes = Partial < Entity > > ( entityClass , attributes , metadataOptions ? ) { if ( isEmptyObject ( attributes ) ) { throw new Error ( "Attributes are required to check it's existence." ) ; } const metadata = this . connection . getEntityByTarget ( entityClass ) ; const uniqueAttributesMetadata = this . connection . getUniqueAttributesForEntity ( entityClass ) ; const uniqueAttributeNames = uniqueAttributesMetadata . map ( attr => attr . name ) ; const { primaryKeyAttributes , uniqueAttributes } = Object . entries ( attributes ) . reduce ( ( acc , [ attrKey , value ] ) => { if ( isUsedForPrimaryKey ( metadata . schema . primaryKey , attrKey ) ) { acc . primaryKeyAttributes [ attrKey ] = value ; } else if ( uniqueAttributeNames . includes ( attrKey ) ) { acc . uniqueAttributes [ attrKey ] = value ; } else { throw new Error ( `Only attributes that are part of primary key or is marked as unique attribute can be queried, attribute "${ attrKey } is neither."` ) ; } return acc ; } , { primaryKeyAttributes : { } as any , uniqueAttributes : { } as any , } ) ; if ( ! isEmptyObject ( primaryKeyAttributes ) && ! isEmptyObject ( uniqueAttributes ) ) { throw new Error ( 'Can not search both primary key and unique attributes at the same time.' ) ; } if ( ! isEmptyObject ( primaryKeyAttributes ) ) { return ! ! ( await this . findOne ( entityClass , attributes , undefined , metadataOptions ) ) ; } if ( ! isEmptyObject ( uniqueAttributes ) ) { const requestId = getUniqueRequestId ( metadataOptions ?. requestId ) ; if ( Object . keys ( uniqueAttributes ) . length > 1 ) { throw new Error ( 'Can only query one unique attribute at a time.' ) ; } const [ attrName , attrValue ] = Object . entries ( uniqueAttributes ) [ 0 ] ; const uniqueAttributePrimaryKey = uniqueAttributesMetadata . find ( meta => meta . name === attrName ) ?. unique ; if ( ! uniqueAttributePrimaryKey ) { console . log ( `Could not find metadata for attribute ${ attrName }` ) ; return false ; } const parsedPrimaryKey = this . _entityTransformer . getParsedPrimaryKey ( metadata . table , uniqueAttributePrimaryKey , { [ attrName ] : attrValue } ) ; const response = await this . connection . documentClient . get ( { Key : { ... parsedPrimaryKey } , TableName : metadata . table . name , ReturnConsumedCapacity : metadataOptions ?. returnConsumedCapacity , } ) . promise ( ) ; if ( response ?. ConsumedCapacity ) { this . connection . logger . logStats ( { requestId , scope : MANAGER_NAME . ENTITY_MANAGER , statsType : STATS_TYPE . CONSUMED_CAPACITY , consumedCapacityData : response . ConsumedCapacity , } ) ; } return ! ! response . Item ; } return false ; } param entityClass param primaryKeyAttributes param body param options async update < Entity , PrimaryKey = Partial < Entity > , AdditionalProperties = Entity > ( entityClass , primaryKeyAttributes , body , options ? , metadataOptions ? ) { const requestId = getUniqueRequestId ( metadataOptions ?. requestId ) ; const dynamoUpdateItem = this . _dcReqTransformer . toDynamoUpdateItem < Entity , PrimaryKey > ( entityClass , primaryKeyAttributes , body , options , { requestId , returnConsumedCapacity : metadataOptions ?. returnConsumedCapacity , } ) ; if ( ! isLazyTransactionWriteItemListLoader ( dynamoUpdateItem ) ) { const response = await this . connection . documentClient . update ( dynamoUpdateItem ) . promise ( ) ; if ( response . ConsumedCapacity ) { this . connection . logger . logStats ( { requestId , scope : MANAGER_NAME . ENTITY_MANAGER , statsType : STATS_TYPE . CONSUMED_CAPACITY , consumedCapacityData : response . ConsumedCapacity , } ) ; } return this . _entityTransformer . fromDynamoEntity < Entity > ( entityClass , response . Attributes ?? { } , { requestId , } ) ; } const existingItem = await this . findOne < Entity , PrimaryKey > ( entityClass , primaryKeyAttributes , undefined , metadataOptions ) ; const updateItemList = dynamoUpdateItem . lazyLoadTransactionWriteItems ( existingItem ) ; await this . connection . transactionManger . writeRaw ( updateItemList , { requestId , returnConsumedCapacity : metadataOptions ?. returnConsumedCapacity , } ) ; return this . findOne < Entity , PrimaryKey > ( entityClass , primaryKeyAttributes , undefined , metadataOptions ) ; } param entityClass param primaryKeyAttributes async delete < Entity , PrimaryKeyAttributes = Partial < Entity > > ( entityClass , primaryKeyAttributes , options ? , metadataOptions ? ) { const requestId = getUniqueRequestId ( metadataOptions ?. requestId ) ; const dynamoDeleteItem = this . _dcReqTransformer . toDynamoDeleteItem < Entity , PrimaryKeyAttributes > ( entityClass , primaryKeyAttributes , options , { requestId , returnConsumedCapacity : metadataOptions ?. returnConsumedCapacity , } ) ; if ( ! isLazyTransactionWriteItemListLoader ( dynamoDeleteItem ) ) { const response = await this . connection . documentClient . delete ( dynamoDeleteItem ) . promise ( ) ; if ( response . ConsumedCapacity ) { this . connection . logger . logStats ( { requestId , scope : MANAGER_NAME . ENTITY_MANAGER , statsType : STATS_TYPE . CONSUMED_CAPACITY , consumedCapacityData : response . ConsumedCapacity , } ) ; } return { success : true , } ; } const existingItem = await this . findOne < Entity , PrimaryKeyAttributes > ( entityClass , primaryKeyAttributes , undefined , metadataOptions ) ; const deleteItemList = dynamoDeleteItem . lazyLoadTransactionWriteItems ( existingItem ) ; await this . connection . transactionManger . writeRaw ( deleteItemList , { requestId , returnConsumedCapacity : metadataOptions ?. returnConsumedCapacity , } ) ; return { success : true , } ; } param entityClass param partitionKey param queryOptions async find < Entity , PartitionKey = Partial < EntityAttributes < Entity > > | string > ( entityClass , partitionKey , queryOptions ? , metadataOptions ? ) { const requestId = getUniqueRequestId ( metadataOptions ?. requestId ) ; const dynamoQueryItem = this . _dcReqTransformer . toDynamoQueryItem < Entity , PartitionKey > ( entityClass , partitionKey , queryOptions , { requestId , returnConsumedCapacity : metadataOptions ?. returnConsumedCapacity , } ) ; const response = await this . _internalRecursiveQuery ( { queryInput : dynamoQueryItem , limit : queryOptions ?. limit ?? getDynamoQueryItemsLimit ( ) , cursor : queryOptions ?. cursor , metadataOptions : { requestId , returnConsumedCapacity : metadataOptions ?. returnConsumedCapacity , } , } ) ; return { ... response , items : response . items . map ( item => this . _entityTransformer . fromDynamoEntity < Entity > ( entityClass , item , { requestId , } ) ) , } ; } param entityClass param partitionKey param queryOptions async count < Entity , PartitionKey = Partial < EntityAttributes < Entity > > | string > ( entityClass , partitionKey , queryOptions ? , metadataOptions ? ) { const requestId = getUniqueRequestId ( metadataOptions ?. requestId ) ; const dynamoQueryItem = this . _dcReqTransformer . toDynamoQueryItem < Entity , PartitionKey > ( entityClass , partitionKey , { ... queryOptions , onlyCount : true , select : undefined } , { requestId , returnConsumedCapacity : metadataOptions ?. returnConsumedCapacity , } ) ; const count = await this . _internalRecursiveCount ( { queryInput : dynamoQueryItem , metadataOptions : { requestId , returnConsumedCapacity : metadataOptions ?. returnConsumedCapacity , } , } ) ; return count ; } param param private async _internalRecursiveQuery ( { queryInput , limit , cursor , itemsFetched = [ ] , metadataOptions , } ) { const { LastEvaluatedKey , Items = [ ] , ConsumedCapacity , } = await this . connection . documentClient . query ( { ... queryInput , ExclusiveStartKey : cursor } ) . promise ( ) ; if ( ConsumedCapacity ) { this . connection . logger . logStats ( { requestId : metadataOptions ?. requestId , scope : MANAGER_NAME . ENTITY_MANAGER , statsType : STATS_TYPE . CONSUMED_CAPACITY , consumedCapacityData : ConsumedCapacity , } ) ; } itemsFetched = [ ... itemsFetched , ... Items ] ; if ( itemsFetched . length < limit && LastEvaluatedKey ) { return this . _internalRecursiveQuery ( { queryInput , limit , cursor : LastEvaluatedKey , itemsFetched , metadataOptions , } ) ; } return { items : itemsFetched , cursor : LastEvaluatedKey } ; } param param private async _internalRecursiveCount ( { queryInput , cursor , currentCount = 0 , metadataOptions , } ) { const { LastEvaluatedKey , Count , ConsumedCapacity , } = await this . connection . documentClient . query ( { ... queryInput , ExclusiveStartKey : cursor } ) . promise ( ) ; if ( ConsumedCapacity ) { this . connection . logger . logStats ( { requestId : metadataOptions ?. requestId , scope : MANAGER_NAME . ENTITY_MANAGER , statsType : STATS_TYPE . CONSUMED_CAPACITY , consumedCapacityData : ConsumedCapacity , } ) ; } currentCount += Count || 0 ; if ( LastEvaluatedKey ) { return this . _internalRecursiveCount ( { queryInput , cursor : LastEvaluatedKey , currentCount , metadataOptions , } ) ; } return currentCount ; } }