import * as estypes from '@elastic/elasticsearch/lib/api/typesWithBodyKey' ; import type { Logger } from '../../../logging' ; import type { SavedObjectsFindOptions , SavedObjectsClientContract } from '../../types' ; import type { SavedObjectsFindResponse } from '../' ; type PointInTimeFinderClient = Pick < SavedObjectsClientContract , 'find' | 'openPointInTimeForType' | 'closePointInTime' > ; public export type SavedObjectsCreatePointInTimeFinderOptions = Omit < SavedObjectsFindOptions , 'page' | 'pit' | 'searchAfter' > ; public export interface SavedObjectsCreatePointInTimeFinderDependencies { client : Pick < SavedObjectsClientContract , 'find' | 'openPointInTimeForType' | 'closePointInTime' > ; } internal export interface PointInTimeFinderDependencies extends SavedObjectsCreatePointInTimeFinderDependencies { logger : Logger ; } internal export type CreatePointInTimeFinderFn = < T = unknown , A = unknown > ( findOptions : SavedObjectsCreatePointInTimeFinderOptions ) => ISavedObjectsPointInTimeFinder < T , A > ; public export interface ISavedObjectsPointInTimeFinder < T , A > { /**
     * An async generator which wraps calls to `savedObjectsClient.find` and
     * iterates over multiple pages of results using `_pit` and `search_after`.
     * This will open a new Point-In-Time (PIT), and continue paging until a set
     * of results is received that's smaller than the designated `perPage` size.
     */ find : ( ) => AsyncGenerator < SavedObjectsFindResponse < T , A > > ; /**
     * Closes the Point-In-Time associated with this finder instance.
     *
     * Once you have retrieved all of the results you need, it is recommended
     * to call `close()` to clean up the PIT and prevent Elasticsearch from
     * consuming resources unnecessarily. This is only required if you are
     * done iterating and have not yet paged through all of the results: the
     * PIT will automatically be closed for you once you reach the last page
     * of results, or if the underlying call to `find` fails for any reason.
     */ close : ( ) => Promise < void > ; } internal export class PointInTimeFinder < T = unknown , A = unknown > implements ISavedObjectsPointInTimeFinder < T , A > { readonly #log : Logger ; readonly #client : PointInTimeFinderClient ; readonly #findOptions : SavedObjectsFindOptions ; #open : boolean = false ; #pitId ? : string ; constructor ( findOptions , { logger , client } ) { this . #log = logger . get ( 'point-in-time-finder' ) ; this . #client = client ; this . #findOptions = { perPage : 1000 , ... findOptions , } ; } async * find ( ) { if ( this . #open ) { throw new Error ( 'Point In Time has already been opened for this finder instance. ' + 'Please call `close()` before calling `find()` again.' ) ; } await this . open ( ) ; let lastResultsCount ; let lastHitSortValue ; do { const results = await this . findNext ( { findOptions : this . #findOptions , id : this . #pitId , searchAfter : lastHitSortValue , } ) ; this . #pitId = results . pit_id ; lastResultsCount = results . saved_objects . length ; lastHitSortValue = this . getLastHitSortValue ( results ) ; this . #log . debug ( `Collected [${ lastResultsCount }] saved objects` ) ; if ( this . #pitId && lastResultsCount < this . #findOptions . perPage ! ) { await this . close ( ) ; } yield results ; } while ( this . #open && lastResultsCount >= this . #findOptions . perPage ! ) ; return ; } async close ( ) { try { if ( this . #pitId ) { this . #log . debug ( `Closing PIT for types [${ this . #findOptions . type }]` ) ; await this . #client . closePointInTime ( this . #pitId ) ; this . #pitId = undefined ; } this . #open = false ; } catch ( e ) { this . #log . error ( `Failed to close PIT for types [${ this . #findOptions . type }]` ) ; throw e ; } } private async open ( ) { try { const { id } = await this . #client . openPointInTimeForType ( this . #findOptions . type , { namespaces : this . #findOptions . namespaces , } ) ; this . #pitId = id ; this . #open = true ; } catch ( e ) { if ( e . output ?. statusCode !== 404 ) { this . #log . error ( `Failed to open PIT for types [${ this . #findOptions . type }]` ) ; throw e ; } this . #log . debug ( `Unable to open PIT for types [${ this . #findOptions . type }]: 404 ${ e }` ) ; } } private async findNext ( { findOptions , id , searchAfter , } ) { try { return await this . #client . find < T , A > ( { sortField : 'updated_at' , sortOrder : 'desc' , pit : id ? { id , keepAlive : '2m' } : undefined , searchAfter , ... findOptions , } ) ; } catch ( e ) { if ( id ) { await this . close ( ) ; } throw e ; } } private getLastHitSortValue ( res ) { if ( res . saved_objects . length < 1 ) { return undefined ; } return res . saved_objects [ res . saved_objects . length - 1 ] . sort ; } }