import { delay } from "awaiting" ; import { once } from "@cocalc/util/async-utils" ; import { SyncTable , SyncTableState } from "@cocalc/sync/table" ; import { TypedMap } from "@cocalc/frontend/app-framework" ; import { close , endswith , merge , path_split , startswith , field_cmp , seconds_ago , } from "@cocalc/util/misc" ; import { DirectoryListingEntry } from "@cocalc/util/types" ; import { get_listing } from "../directory-listing" ; import { WATCH_TIMEOUT_MS , MAX_FILES_PER_PATH , } from "@cocalc/util/db-schema/listings" ; import { Watcher } from "./path-watcher" ; import { close_all_syncdocs_in_tree } from "./sync-doc" ; import { remove_jupyter_backend } from "../jupyter/jupyter" ; const WATCH_DEBOUNCE_MS = 1500 ; const DELAY_ON_CHANGE_MS = 50 ; const INTEREST_THRESH_SECONDS = WATCH_TIMEOUT_MS / 1000 ; import { MAX_PATHS } from "@cocalc/util/db-schema/listings" ; interface Listing { path : string ; project_id ? : string ; listing ? : DirectoryListingEntry [ ] ; time ? : Date ; interest ? : Date ; missing ? : number ; error ? : string ; deleted ? : string [ ] ; } export type ImmutableListing = TypedMap < Listing > ; class ListingsTable { private readonly table ? : SyncTable ; private logger : undefined | { debug : Function ; } ; private project_id : string ; private watchers : { [ path ] : Watcher ; } = { } ; constructor ( table , logger , project_id ) { this . project_id = project_id ; this . logger = logger ; this . log ( "register" ) ; this . table = table ; this . setup_watchers ( ) ; } public close ( ) { this . log ( "close" ) ; for ( const path in this . watchers ) { this . stop_watching ( path ) ; } close ( this ) ; } private async setup_watchers ( ) { if ( this . table == null ) return ; if ( this . table . get_state ( ) == ( "init" as SyncTableState ) ) { await once ( this . table , "state" ) ; } if ( this . table . get_state ( ) != ( "connected" as SyncTableState ) ) { return ; } this . table . get ( ) ?. forEach ( ( val ) => { const path = val . get ( "path" ) ; if ( path == null ) return ; if ( this . watchers [ path ] == null ) return ; const interest = val . get ( "interest" ) ; if ( interest != null && interest > seconds_ago ( INTEREST_THRESH_SECONDS ) ) { this . start_watching ( path ) ; } } ) ; this . table . on ( "change" , this . handle_change_event . bind ( this ) ) ; this . remove_stale_watchers ( ) ; } private async remove_stale_watchers ( ) { if ( this . table == null ) return ; if ( this . table . get_state ( ) == ( "connected" as SyncTableState ) ) { this . table . get ( ) ?. forEach ( ( val ) => { const path = val . get ( "path" ) ; if ( path == null ) return ; if ( this . watchers [ path ] == null ) return ; const interest = val . get ( "interest" ) ; if ( interest == null || interest <= seconds_ago ( INTEREST_THRESH_SECONDS ) ) { this . stop_watching ( path ) ; } } ) ; } try { await this . trim_old_paths ( ) ; } catch ( err ) { this . log ( "WARNING, error trimming old paths -- " , err ) ; } if ( this . table == null ) return ; if ( this . table . get_state ( ) == ( "connected" as SyncTableState ) ) { await delay ( 1000 * INTEREST_THRESH_SECONDS ) ; if ( this . table == null ) return ; if ( this . table . get_state ( ) != ( "connected" as SyncTableState ) ) return ; this . remove_stale_watchers ( ) ; } } private log ( ... args ) { if ( this . logger == null ) return ; this . logger . debug ( "listings_table" , ... args ) ; } private is_ready ( ) { return ! ! this . table ?. is_ready ( ) ; } private get_table ( ) { if ( ! this . is_ready ( ) || this . table == null ) { throw Error ( "table not ready" ) ; } return this . table ; } async set ( obj ) { this . get_table ( ) . set ( merge ( { project_id : this . project_id } , obj ) , "shallow" ) ; await this . get_table ( ) . save ( ) ; } public get ( path ) { const x = this . get_table ( ) . get ( JSON . stringify ( [ this . project_id , path ] ) ) ; if ( x == null ) return x ; return ( x as unknown ) as ImmutableListing ; } private handle_change_event ( keys ) { this . log ( "handle_change_event" , JSON . stringify ( keys ) ) ; for ( const key of keys ) { this . handle_change ( JSON . parse ( key ) [ 1 ] ) ; } } private handle_change ( path ) { this . log ( "handle_change" , path ) ; const cur = this . get ( path ) ; if ( cur == null ) return ; let interest = cur . get ( "interest" ) ; if ( interest == null ) return ; if ( interest >= seconds_ago ( INTEREST_THRESH_SECONDS ) ) { const time = new Date ( ) ; if ( interest > time ) { interest = time ; this . set ( { path , interest } ) ; } this . ensure_watching ( path ) ; } } private async ensure_watching ( path ) { if ( this . watchers [ path ] != null ) { return ; } try { await this . compute_listing ( path ) ; } catch ( err ) { this . log ( "ensure_watching -- failed to compute listing so not starting watching" , err ) ; return ; } try { this . start_watching ( path ) ; } catch ( err ) { this . log ( "failed to start watching" , err ) ; } } private async compute_listing ( path ) { const time = new Date ( ) ; let listing ; try { listing = await get_listing ( path , true ) ; if ( ! this . is_ready ( ) ) return ; } catch ( err ) { if ( ! this . is_ready ( ) ) return ; this . set ( { path , time , error : `${ err }` } ) ; throw err ; } let missing = undefined ; const y = this . get ( path ) ; const previous_listing = y ?. get ( "listing" ) ?. toJS ( ) ; let deleted = y ?. get ( "deleted" ) ?. toJS ( ) ; if ( previous_listing != null ) { const cur = new Set ( ) ; for ( const x of listing ) { cur . add ( x . name ) ; } for ( const x of previous_listing ) { if ( ! cur . has ( x . name ) ) { if ( deleted == null ) { deleted = [ x . name ] ; } else { if ( deleted . indexOf ( x . name ) == - 1 ) { deleted . push ( x . name ) ; } } } } } if ( listing . length > MAX_FILES_PER_PATH ) { listing . sort ( field_cmp ( "mtime" ) ) ; listing . reverse ( ) ; missing = listing . length - MAX_FILES_PER_PATH ; listing = listing . slice ( 0 , MAX_FILES_PER_PATH ) ; } const error = y ?. get ( "error" ) != null ? "" : undefined ; this . set ( { path , listing , time , missing , deleted , error } ) ; } private start_watching ( path ) { if ( this . watchers [ path ] != null ) return ; if ( process . env . HOME == null ) { throw Error ( "HOME env variable must be defined" ) ; } this . watchers [ path ] = new Watcher ( path , WATCH_DEBOUNCE_MS , this . log . bind ( this ) ) ; this . watchers [ path ] . on ( "change" , async ( ) => { try { await delay ( DELAY_ON_CHANGE_MS ) ; if ( ! this . is_ready ( ) ) return ; await this . compute_listing ( path ) ; } catch ( err ) { this . log ( `compute_listing("${ path }") error: "${ err }"` ) ; } } ) ; } private stop_watching ( path ) { const w = this . watchers [ path ] ; if ( w == null ) return ; delete this . watchers [ path ] ; w . close ( ) ; } private async trim_old_paths ( ) { this . log ( "trim_old_paths" ) ; if ( ! this . is_ready ( ) ) return ; const table = this . get_table ( ) ; let num_to_remove = table . size ( ) - MAX_PATHS ; this . log ( "trim_old_paths" , num_to_remove ) ; if ( num_to_remove <= 0 ) { return ; } const paths = [ ] ; table . get ( ) ?. forEach ( ( val ) => { const path = val . get ( "path" ) ; if ( this . watchers [ path ] != null ) { num_to_remove -= 1 ; return ; } const interest = val . get ( "interest" , new Date ( 0 ) ) ; paths . push ( { path , interest } ) ; } ) ; this . log ( "trim_old_paths" , JSON . stringify ( paths ) ) ; this . log ( "trim_old_paths" , num_to_remove ) ; if ( num_to_remove <= 0 ) return ; paths . sort ( field_cmp ( "interest" ) ) ; for ( let i = 0 ; i < num_to_remove ; i ++ ) { this . log ( "trim_old_paths -- removing" , paths [ i ] . path ) ; await this . remove_path ( paths [ i ] . path ) ; } } private async remove_path ( path ) { if ( ! this . is_ready ( ) ) return ; this . log ( "remove_path" , path ) ; await this . get_table ( ) . delete ( { project_id : this . project_id , path } ) ; } public async set_deleted ( filename ) { this . log ( `set_deleted: ${ filename }` ) ; if ( ! this . is_ready ( ) ) { this . log ( `set_deleted: skipping since not ready` ) ; return ; } if ( filename [ 0 ] == "/" ) { if ( process . env . HOME == null || ! startswith ( filename , process . env . HOME ) ) { return ; } filename = filename . slice ( process . env . HOME . length + 1 ) ; } const { head , tail } = path_split ( filename ) ; const x = this . get ( head ) ; if ( x != null ) { let deleted = x . get ( "deleted" ) ; if ( deleted == null ) { deleted = [ tail ] ; } else { if ( deleted . indexOf ( tail ) != - 1 ) return ; deleted = deleted . toJS ( ) ; deleted . push ( tail ) ; } this . log ( `set_deleted: recording "${ deleted }" in "${ head }"` ) ; await this . set ( { path : head , deleted } ) ; if ( ! this . is_ready ( ) ) return ; } await close_all_syncdocs_in_tree ( filename ) ; if ( ! this . is_ready ( ) ) return ; if ( endswith ( filename , ".ipynb" ) ) { this . log ( `set_deleted: handling jupyter kernel for ${ filename }` ) ; await remove_jupyter_backend ( filename , this . project_id ) ; if ( ! this . is_ready ( ) ) return ; } } public is_deleted ( filename ) { if ( ! this . is_ready ( ) ) { return false ; } const { head , tail } = path_split ( filename ) ; if ( head != "" && this . is_deleted ( head ) ) { return true ; } const x = this . get ( head ) ; if ( x == null ) { return false ; } const deleted = x . get ( "deleted" ) ; if ( deleted == null ) { return false ; } return deleted . indexOf ( tail ) != - 1 ; } } let listings_table = undefined ; export function register_listings_table ( table , logger , project_id ) { logger . debug ( "register_listings_table" ) ; if ( listings_table != null ) { listings_table . close ( ) ; } listings_table = new ListingsTable ( table , logger , project_id ) ; return ; } export function get_listings_table ( ) { return listings_table ; }