import Debug from '@prisma/debug' ; import type { MigrateEngineLogLine } from '@prisma/sdk' ; import { BinaryType , ErrorArea , resolveBinary , RustPanic , MigrateEngineExitCode } from '@prisma/sdk' ; import chalk from 'chalk' ; import type { ChildProcess } from 'child_process' ; import { spawn } from 'child_process' ; import type { EngineArgs , EngineResults } from './types' ; import byline from './utils/byline' ; const debugRpc = Debug ( 'prisma:migrateEngine:rpc' ) ; const debugStderr = Debug ( 'prisma:migrateEngine:stderr' ) ; const debugStdin = Debug ( 'prisma:migrateEngine:stdin' ) ; export interface MigrateEngineOptions { projectDir : string ; schemaPath : string ; debug ? : boolean ; enabledPreviewFeatures ? : string [ ] ; } export interface RPCPayload { id : number ; jsonrpc : string ; method : string ; params : any ; } export class EngineError extends Error { public code : number ; constructor ( message , code ) { super ( message ) ; this . code = code ; } } let messageId = 1 ; export class MigrateEngine { private projectDir : string ; private debug : boolean ; private child ? : ChildProcess ; private schemaPath : string ; private listeners : { [ key ] : ( result : any , err ? : any ) => any ; } = { } ; /**  _All_ the logs from the engine process. */ private messages : string [ ] = [ ] ; private lastRequest ? : any ; /** The fields of the last engine log event with an `ERROR` level. */ private lastError : MigrateEngineLogLine [ 'fields' ] | null = null ; private initPromise ? : Promise < void > ; private enabledPreviewFeatures ? : string [ ] ; constructor ( { projectDir , debug = false , schemaPath , enabledPreviewFeatures } ) { this . projectDir = projectDir ; this . schemaPath = schemaPath ; if ( debug ) { Debug . enable ( 'MigrateEngine*' ) ; } this . debug = debug ; this . enabledPreviewFeatures = enabledPreviewFeatures ; } public stop ( ) { this . child ! . kill ( ) ; } public devDiagnostic ( args ) { return this . runCommand ( this . getRPCPayload ( 'devDiagnostic' , args ) ) ; } public listMigrationDirectories ( args ) { return this . runCommand ( this . getRPCPayload ( 'listMigrationDirectories' , args ) ) ; } public markMigrationApplied ( args ) { return this . runCommand ( this . getRPCPayload ( 'markMigrationApplied' , args ) ) ; } public markMigrationRolledBack ( args ) { return this . runCommand ( this . getRPCPayload ( 'markMigrationRolledBack' , args ) ) ; } public diagnoseMigrationHistory ( args ) { return this . runCommand ( this . getRPCPayload ( 'diagnoseMigrationHistory' , args ) ) ; } public planMigration ( args ) { return this . runCommand ( this . getRPCPayload ( 'planMigration' , args ) ) ; } public evaluateDataLoss ( args ) { return this . runCommand ( this . getRPCPayload ( 'evaluateDataLoss' , args ) ) ; } public createMigration ( args ) { return this . runCommand ( this . getRPCPayload ( 'createMigration' , args ) ) ; } public applyMigrations ( args ) { return this . runCommand ( this . getRPCPayload ( 'applyMigrations' , args ) ) ; } public reset ( ) { return this . runCommand ( this . getRPCPayload ( 'reset' , undefined ) ) ; } public getDatabaseVersion ( ) { return this . runCommand ( this . getRPCPayload ( 'getDatabaseVersion' , undefined ) ) ; } public schemaPush ( args ) { return this . runCommand ( this . getRPCPayload ( 'schemaPush' , args ) ) ; } public debugPanic ( ) { return this . runCommand ( this . getRPCPayload ( 'debugPanic' , undefined ) ) ; } private rejectAll ( err ) { Object . entries ( this . listeners ) . map ( ( [ id , listener ] ) => { listener ( null , err ) ; delete this . listeners [ id ] ; } ) ; } private registerCallback ( id , callback ) { this . listeners [ id ] = callback ; } private handleResponse ( response ) { let result ; try { result = JSON . parse ( response ) ; } catch ( e ) { console . error ( `Could not parse migration engine response: ${ response . slice ( 0 , 200 ) }` ) ; } if ( result ) { if ( result . id ) { if ( ! this . listeners [ result . id ] ) { console . error ( `Got result for unknown id ${ result . id }` ) ; } if ( this . listeners [ result . id ] ) { this . listeners [ result . id ] ( result ) ; delete this . listeners [ result . id ] ; } } else { if ( result . is_panic ) { throw new Error ( `Response: ${ result . message }` ) ; } else if ( result . message ) { console . error ( chalk . red ( `Response: ${ result . message }` ) ) ; } else { console . error ( chalk . red ( `Response: ${ JSON . stringify ( result ) }` ) ) ; } } } } private init ( ) { if ( ! this . initPromise ) { this . initPromise = this . internalInit ( ) ; } return this . initPromise ; } private internalInit ( ) { return new Promise ( async ( resolve , reject ) => { try { const { PWD , ... rest } = process . env ; const binaryPath = await resolveBinary ( BinaryType . migrationEngine ) ; debugRpc ( 'starting migration engine with binary: ' + binaryPath ) ; const args = [ '-d' , this . schemaPath ] ; if ( this . enabledPreviewFeatures && Array . isArray ( this . enabledPreviewFeatures ) && this . enabledPreviewFeatures . length > 0 ) { args . push ( ... [ '--enabled-preview-features' , this . enabledPreviewFeatures . join ( ',' ) ] ) ; } this . child = spawn ( binaryPath , args , { cwd : this . projectDir , stdio : [ 'pipe' , 'pipe' , this . debug ? process . stderr : 'pipe' ] , env : { ... rest , SERVER_ROOT : this . projectDir , RUST_LOG : 'info' , RUST_BACKTRACE : '1' , } , } ) ; this . child . on ( 'error' , ( err ) => { console . error ( '[migration-engine] error: %s' , err ) ; this . rejectAll ( err ) ; reject ( err ) ; } ) ; this . child . on ( 'exit' , ( code ) : void => { const exitWithErr = ( err ) : void => { this . rejectAll ( err ) ; reject ( err ) ; } ; const engineMessage = this . lastError ?. message || this . messages . join ( '\n' ) ; const handlePanic = ( ) => { const stackTrace = this . messages . join ( '\n' ) ; exitWithErr ( new RustPanic ( serializePanic ( engineMessage ) , stackTrace , this . lastRequest , ErrorArea . LIFT_CLI , this . schemaPath ) ) ; } ; switch ( code ) { case MigrateEngineExitCode . Success : break ; case MigrateEngineExitCode . Error : exitWithErr ( new Error ( `Error in migration engine: ${ engineMessage }` ) ) ; break ; case MigrateEngineExitCode . Panic : handlePanic ( ) ; break ; default : handlePanic ( ) ; } } ) ; this . child . stdin ! . on ( 'error' , ( err ) => { debugStdin ( err ) ; } ) ; byline ( this . child . stderr ) . on ( 'data' , ( msg ) => { const data = String ( msg ) ; debugStderr ( data ) ; try { const json = JSON . parse ( data ) ; if ( json . fields ?. migrate_action === 'log' ) { console . info ( json . fields . message ) ; } this . messages . push ( json . fields . message ) ; if ( json . level === 'ERROR' ) { this . lastError = json . fields ; } } catch ( e ) { } } ) ; byline ( this . child . stdout ) . on ( 'data' , ( line ) => { this . handleResponse ( String ( line ) ) ; } ) ; setImmediate ( ( ) => { resolve ( ) ; } ) ; } catch ( e ) { reject ( e ) ; } } ) ; } private async runCommand ( request ) { if ( process . env . FORCE_PANIC_MIGRATION_ENGINE ) { request = this . getRPCPayload ( 'debugPanic' , undefined ) ; } await this . init ( ) ; if ( this . child ?. killed ) { throw new Error ( `Can't execute ${ JSON . stringify ( request ) } because migration engine already exited.` ) ; } return new Promise ( ( resolve , reject ) => { this . registerCallback ( request . id , ( response , err ) => { if ( err ) { return reject ( err ) ; } if ( response . result !== undefined ) { resolve ( response . result ) ; } else { if ( response . error ) { debugRpc ( response ) ; if ( response . error . data ?. is_panic ) { const message = response . error . data ?. error ?. message ?? response . error . message ; reject ( new RustPanic ( message , response . error . data . message , this . lastRequest , ErrorArea . LIFT_CLI , this . schemaPath ) ) ; } else if ( response . error . data ?. message ) { let message = `${ chalk . redBright ( response . error . data . message ) }\n` ; if ( response . error . data ?. error_code ) { message = chalk . redBright ( `${ response . error . data . error_code }\n\n` ) + message ; reject ( new EngineError ( message , response . error . data . error_code ) ) ; } else { reject ( new Error ( message ) ) ; } } else { reject ( new Error ( `${ chalk . redBright ( 'Error in RPC' ) }\n Request: ${ JSON . stringify ( request , null , 2 ) }\nResponse: ${ JSON . stringify ( response , null , 2 ) }\n${ response . error . message }\n` ) ) ; } } else { reject ( new Error ( `Got invalid RPC response without .result property: ${ JSON . stringify ( response ) }` ) ) ; } } } ) ; if ( this . child ! . stdin ! . destroyed ) { throw new Error ( `Can't execute ${ JSON . stringify ( request ) } because migration engine is destroyed.` ) ; } debugRpc ( 'SENDING RPC CALL' , JSON . stringify ( request ) ) ; this . child ! . stdin ! . write ( JSON . stringify ( request ) + '\n' ) ; this . lastRequest = request ; } ) ; } private getRPCPayload ( method , params ) { return { id : messageId ++ , jsonrpc : '2.0' , method , params : { ... params , } , } ; } } /** The full message with context we return to the user in case of engine panic. */ function serializePanic ( log ) { return `${ chalk . red . bold ( 'Error in migration engine.\nReason: ' ) }${ log }

Please create an issue with your \`schema.prisma\` at
${ chalk . underline ( 'https://github.com/prisma/prisma/issues/new' ) }\n` ; }