import { UserError } from "@adpt/utils" ; import AsyncLock from "async-lock" ; import pDefer from "p-defer" ; import { InternalError } from "../error" ; import { $serverLock , OptionsWithLock , ServerLock , } from "./server" ; export interface Locker < LockType > { lock ( ) : Promise < LockType > ; unlock ( lock ) : Promise < void > ; } export interface NoLock extends ServerLock { } export class NoLocker implements Locker < NoLock > { lock ( ) { return Promise . resolve < NoLock > ( { [ $serverLock ] : true } ) ; } unlock ( l ) { return Promise . resolve ( ) ; } } export interface ProcessLock extends ServerLock { release : ( ) => Promise < void > ; } export class ProcessLocker implements Locker < ProcessLock > { protected locks = new AsyncLock ( ) ; lock ( ) { return new Promise < ProcessLock > ( ( resolve , reject ) => { const releaseComplete = pDefer < void > ( ) ; this . locks . acquire ( "thelock" , ( releaseLock ) => { resolve ( { release : async ( ) => { releaseLock ( ) ; return releaseComplete . promise ; } , [ $serverLock ] : true , } ) ; } , ( err ) => { if ( err ) { reject ( err ) ; releaseComplete . reject ( err ) ; } else { releaseComplete . resolve ( ) ; } } ) ; } ) ; } async unlock ( lock ) { await lock . release ( ) ; } } export abstract class ServerBase < L extends ServerLock = ServerLock > { protected processLocker = new ProcessLocker ( ) ; protected worldLocker : Locker < L > ; protected currentProcessLock ? : ProcessLock ; protected currentWorldLock ? : L ; constructor ( worldLocker ? ) { this . worldLocker = worldLocker || ( new NoLocker ( ) ) as Locker < L > ; } async lock ( ) { let pLock ; try { pLock = await this . processLocker . lock ( ) ; if ( this . currentProcessLock ) { throw new InternalError ( `AdaptServer: previous process lock not cleared` ) ; } this . currentProcessLock = pLock ; try { this . currentWorldLock = await this . worldLocker . lock ( ) ; } catch ( e ) { throw new UserError ( `Unable to get exclusive access to Adapt server. ` + `Please retry in a moment. [${ e . message }]` ) ; } return pLock ; } catch ( err ) { if ( pLock ) { if ( pLock === this . currentProcessLock ) delete this . currentProcessLock ; await this . processLocker . unlock ( pLock ) ; } throw err ; } } async unlock ( l ) { if ( ! this . assertLock ( l , "unlock" ) ) throw new InternalError ( `Bad lock` ) ; const wLock = this . currentWorldLock ; if ( wLock ) { delete this . currentWorldLock ; await this . worldLocker . unlock ( wLock ) ; } delete this . currentProcessLock ; await this . processLocker . unlock ( l ) ; } protected async withLock < T > ( options , f ) { let release ; if ( options . lock ) { this . assertLock ( options . lock , "use" ) ; release = async ( ) => { } ; } else { const tempLock = await this . lock ( ) ; release = ( ) => this . unlock ( tempLock ) ; } try { return await f ( ) ; } finally { await release ( ) ; } } protected assertLock ( l , op ) { if ( ! this . currentProcessLock ) { throw new InternalError ( `AdaptServer: Attempt to ${ op } a stale lock. Server is not ` + `currently locked.` ) ; } if ( l !== this . currentProcessLock ) { throw new InternalError ( `AdaptServer: Attempt to ${ op } a stale lock. Server is locked ` + `by another user.` ) ; } return true ; } }