import * as TaskEither from 'fp-ts/lib/TaskEither' ; import * as Either from 'fp-ts/lib/Either' ; import { SavedObjectSanitizedDoc , SavedObjectsRawDoc , SavedObjectsSerializer , SavedObjectUnsanitizedDoc , } from '../../serialization' ; import { MigrateAndConvertFn } from './document_migrator' ; import { TransformSavedObjectDocumentError } from '.' ; export interface DocumentsTransformFailed { readonly type : string ; readonly corruptDocumentIds : string [ ] ; readonly transformErrors : TransformErrorObjects [ ] ; } export interface DocumentsTransformSuccess { readonly processedDocs : SavedObjectsRawDoc [ ] ; } export interface TransformErrorObjects { readonly rawId : string ; readonly err : TransformSavedObjectDocumentError | Error ; } type MigrateFn = ( doc : SavedObjectUnsanitizedDoc < unknown > ) => Promise < Array < SavedObjectUnsanitizedDoc < unknown > > > ; /**
 * Error thrown when saved object migrations encounter a corrupt saved object.
 * Corrupt saved objects cannot be serialized because:
 *  - there's no `[type]` property which contains the type attributes
 *  - the type or namespace in the _id doesn't match the `type` or `namespace`
 *    properties
 */ export class CorruptSavedObjectError extends Error { constructor ( public readonly rawId ) { super ( `Unable to migrate the corrupt saved object document with _id: '${ rawId }'.` ) ; Object . setPrototypeOf ( this , CorruptSavedObjectError . prototype ) ; } } param { TransformFn } migrateDoc param { SavedObjectsRawDoc [ ] } rawDocs returns { SavedObjectsRawDoc [ ] } export async function migrateRawDocs ( serializer , migrateDoc , rawDocs ) { const migrateDocWithoutBlocking = transformNonBlocking ( migrateDoc ) ; const processedDocs = [ ] ; for ( const raw of rawDocs ) { const options = { namespaceTreatment : 'lax' as const } ; if ( serializer . isRawSavedObject ( raw , options ) ) { const savedObject = convertToRawAddMigrationVersion ( raw , options , serializer ) ; processedDocs . push ( ... ( await migrateMapToRawDoc ( migrateDocWithoutBlocking , savedObject , serializer ) ) ) ; } else { throw new CorruptSavedObjectError ( raw . _id ) ; } } return processedDocs ; } interface MigrateRawDocsSafelyDeps { serializer : SavedObjectsSerializer ; migrateDoc : MigrateAndConvertFn ; rawDocs : SavedObjectsRawDoc [ ] ; } returns { TaskEither . TaskEither < DocumentsTransformFailed , DocumentsTransformSuccess > } export function migrateRawDocsSafely ( { serializer , migrateDoc , rawDocs , } ) { return async ( ) => { const migrateDocNonBlocking = transformNonBlocking ( migrateDoc ) ; const processedDocs = [ ] ; const transformErrors = [ ] ; const corruptSavedObjectIds = [ ] ; const options = { namespaceTreatment : 'lax' as const } ; for ( const raw of rawDocs ) { if ( serializer . isRawSavedObject ( raw , options ) ) { try { const savedObject = convertToRawAddMigrationVersion ( raw , options , serializer ) ; processedDocs . push ( ... ( await migrateMapToRawDoc ( migrateDocNonBlocking , savedObject , serializer ) ) ) ; } catch ( err ) { if ( err instanceof TransformSavedObjectDocumentError ) { transformErrors . push ( { rawId : raw . _id , err , } ) ; } else { transformErrors . push ( { rawId : raw . _id , err } ) ; } } } else { corruptSavedObjectIds . push ( raw . _id ) ; } } if ( corruptSavedObjectIds . length > 0 || transformErrors . length > 0 ) { return Either . left ( { type : 'documents_transform_failed' , corruptDocumentIds : [ ... corruptSavedObjectIds ] , transformErrors , } ) ; } return Either . right ( { processedDocs } ) ; } ; } /**
 * Migration transform functions are potentially CPU heavy e.g. doing decryption/encryption
 * or (de)/serializing large JSON payloads.
 * Executing all transforms for a batch in a synchronous loop can block the event-loop for a long time.
 * To prevent this we use setImmediate to ensure that the event-loop can process other parallel
 * work in between each transform.
 */ function transformNonBlocking ( transform ) { return ( doc ) => new Promise ( ( resolve , reject ) => { setImmediate ( ( ) => { try { resolve ( transform ( doc ) ) ; } catch ( e ) { reject ( e ) ; } } ) ; } ) ; } param { MigrateFn } transformNonBlocking param { SavedObjectsRawDoc [ ] } rawDoc returns { Promise < SavedObjectsRawDoc [ ] > } async function migrateMapToRawDoc ( migrateMethod , savedObject , serializer ) { return [ ... ( await migrateMethod ( savedObject ) ) ] . map ( ( attrs ) => serializer . savedObjectToRaw ( { references : [ ] , ... attrs , } ) ) ; } param { SavedObjectRawDoc } rawDoc param options param { SavedObjectsSerializer } serializer returns { SavedObjectSanitizedDoc < unknown > } function convertToRawAddMigrationVersion ( rawDoc , options , serializer ) { const savedObject = serializer . rawToSavedObject ( rawDoc , options ) ; savedObject . migrationVersion = savedObject . migrationVersion || { } ; return savedObject ; }