batch fetch

batch.ts

namespace $ {
	type Unit<Full, Patch> = $mol_promise<Full | null> & {
		id: string
		patch?: Patch | null
		error: Error
	}

	export class $gd_kit_batch<
		Full extends {},
		Patch extends ({} | null) = Partial<Full>,
	> extends $mol_object {
		protected sync_timeout = undefined as undefined | $mol_after_timeout

		id() {
			return ''
		}

		destructor() {
			this.units = []
			this.sync_timeout?.destructor()
			this.sync_timeout = undefined
		}

		@ $mol_action
		load(ids: readonly string[]): Record<string, Full> {
			throw new Error('implement')
		}

		@ $mol_action
		patch(patches: Record<string, Patch | null>): Record<string, Full> {
			throw new Error('implement')
		}

		strict_ids_check() {
			return false
		}

		protected units = [] as Unit<Full, Patch>[]

		protected async batch() {
			this.sync_timeout = undefined

			const units = this.units
			if( units.length === 0 ) return
			this.units = []

			const load_ids = new Set<string>()
			const patches = {} as Record<string, Patch | null>

			let has_patches = false

			for (const { id, patch } of units) {
				if (patch === undefined) {
					load_ids.add(id)
					continue
				}

				patches[id] = patch === null ? null : Object.assign(patches[id] ?? {}, patch)
				has_patches = true
			}

			let not_fetched = units

			try {
				const [ loaded, patched ] = await Promise.all([
					load_ids.size ? $mol_wire_async(this).load([ ...load_ids ]) : undefined,
					has_patches ? $mol_wire_async(this).patch(patches) : undefined,
				])

				const strict_check = this.strict_ids_check()
	
				not_fetched = []
	
				for( const unit of units ) {
					const id = unit.id
					const val = patched?.[id] ?? loaded?.[id] ?? null
					if (! val && strict_check) not_fetched.push(unit)
					else unit.done(val)
				}

				if (not_fetched.length) {
					throw new Error(`Not fetched ids`, { cause: { ids: not_fetched.map(unit => unit.id) } })
				}
			} catch (e) {
				for( const unit of not_fetched) unit.fail(new $mol_error_mix(
					(e as Error).message || 'Не сохранился объект',
					{ id: unit.id, patch: unit.patch },
					e as Error,
					unit.error,
				))
			}
		}

		timeout() {
			return 70
		}

		static place = undefined as undefined | string

		data_async(id: string, patch?: Patch | null | false) {
			const prev = this.units.find(u => u.id === id)
			if (prev && patch) prev.patch = { ...prev.patch, ...patch }
			if (prev) return prev
			const error = new Error('Stack save') // new Error() сохраняет стек трейс для отладки
			const unit: Unit<Full, Patch> = Object.assign(new $mol_promise<Full | null>(), { id, patch: patch === false ? undefined : patch, error })
			this.units.push(unit)

			if (this.sync_timeout) return unit


			this.sync_timeout = new this.$.$mol_after_timeout(
				this.timeout(),
				this.batch.bind(this)
			)
			this.$.$gd_kit_batch.place = undefined

			return unit
		}

		cache(id: string, cache?: Full | Error | null) {
			return this.data(id, null, cache)
		}

		static level = 0
		static promises = {} as Record<string, Unit<any, any>>
		static wait_async() {
			return Promise.all(Object.values(this.promises))
		}

		@ $mol_action
		static wait(id: string) {
			return this.promises[id] ?? null
		}

		static transaction<V>(task: () => V): V {
			try {
				const prev_level = this.level
				this.level++

				const result = task()

				if (prev_level === 0) {
					$mol_wire_sync(this).wait_async()
				}
				this.promises = {}

				return result

			} catch(e) {
				if (! $mol_promise_like(e)) this.promises = {}
				$mol_fail_hidden(e)
			} finally {
				this.level--
			}
		}

		private _cache = {} as Record<string, Full | undefined | null>

		cache_dto(id: string, next?: Full | null) {
			if (next !== undefined) this._cache[id] = next
			return this._cache[id]
		}

		@ $mol_mem_key
		data(id: string, patch?: Patch | null | false, cache?: Full | Error | null): Full | null {
			// $mol_wire_solid()
			if (cache) {
				this.cache_dto(id, cache instanceof Error ? undefined : cache)
				this.$.$gd_kit_batch.promises[id]?.done(cache)
				return cache as Full
			}

			if (patch && this.$.$gd_kit_batch.level > 0) {
				const unit = this.data_async(id, patch)
				const set = this.data.bind(this, id, null)
				const promise = unit.then(set).catch(set)

				this.$.$gd_kit_batch.promises[id] = Object.assign(promise, unit)

				const prev = $mol_wire_probe(() => this.data(id))

				const res = {
					...prev,
					...patch as unknown as Full
				}
				this.cache_dto(id, res)

				return res
			}

			const cached = this.cache_dto(id)
			if (cached !== undefined && patch === undefined) return cached
			const result = $mol_wire_sync(this).data_async(id, patch)
			this.cache_dto(id, result)
			return result
		}

	}
}

form.ts

namespace $.$$ {
	export class $gd_kit_form extends $.$gd_kit_form {
		@ $mol_action
		override submit() {
			this.$.$gd_kit_batch.transaction(() => super.submit())
			this.result( this.message_done() )
		}

	}
}

registry.ts

namespace $ {

	// Работа с множеством проп-ов в базе, загрузка и денормализация
	export class $gd_kit_object_registry extends $mol_object {
		protected init() {
			return this.$.$gd_kit_init_main
		}

		@ $mol_mem
		static meta<Instance>( this: { new(): Instance } ) {
			return ( this as typeof this & { make: ( config: Partial<$gd_kit_object_registry> ) => Instance } ).make( {
				data_type: () => 'meta',
			} )
		}

		@ $mol_mem
		static big<Instance>( this: { new(): Instance } ) {
			return ( this as typeof this & { make: ( config: Partial<$gd_kit_object_registry> ) => Instance } ).make( {
				data_type: () => 'big',
			} )
		}

		core_url() {
			return this.init().service_core_url()
		}

		ws_url() {
			this.$.$gd_kit_session.token() // restart ws if token changed
			return this.$.$gd_kit_transport.base_url_ws() + this.core_url() + 'ws/sub'
		}

		protected ws_db() {
			return this.data_type() === 'big'
				? this.init().wbd_id()
				: this.init().db_current_name()
		}

		@ $mol_mem
		ws() {
			return this.$.$gd_kit_object_ws.make( {
				url: () => this.ws_url(),
				token: () => this.token(),
				init: () => this.ws_auth(),
				id_prefix: () => this.ws_db(),
				pg_cmd_ins: obj => this.ws_pg_cmd_ins( obj ),
				pg_cmd_upd: obj => this.ws_pg_cmd_ins( obj, false, true ),

			} )
		}

		protected token() {
			const logged = this.$.$gd_kit_session.logged()
			const token = this.$.$gd_kit_transport.token()
			if( !logged || !token ) return null

			return token
		}

		@ $mol_action
		protected ws_auth() {
			const token = this.token()
			if (! token) return

			const db = this.ws_db()
			this.ws().send_object( { type: 'auth', data: { token, db } } )

			this.$.$mol_log3_rise( {
				place: '$gd_kit_object_registry.ws_auth()',
				socket_id: this.ws().id(),
				message: 'socket auth',
			} )
		}

		ws_status() {
			return this.ws().status()
		}

		ws_error() {
			return this.ws().error_last()
		}

		protected pg_inserts = [] as typeof $gd_kit_prop_data.Value[]

		protected ws_pg_cmd_ins( prop: typeof $gd_kit_prop_data.Value, replay = false, update = false ) {

			const matched_first = Boolean( $mol_wire_probe( () => this.batch_by_p()?.data( prop.p ) ) ) // проверка, есть ли prop.p в кэше

			let matched = matched_first
			if( matched ) this.cache( prop )

			for( const search of this.searches ) {
				if( !search.matched( prop ) ) continue
				matched = true
				this.cache( prop )
				if (! matched_first) search.insert( prop.p )
			}

			if( !replay && !matched ) this.pg_inserts.push( prop )
		}

		protected replay_timer: undefined | $mol_after_timeout = undefined

		replay_delay() {
			return 10
		}

		replay() {
			while( this.pg_inserts.length ) {
				try {
					this.ws_pg_cmd_ins( this.pg_inserts[ 0 ], true )
				} catch( e ) {
					if( $mol_promise_like( e ) ) $mol_fail_hidden( e )

					this.$.$mol_log3_fail( {
						place: '$gd_kit_object_registry.replay()',
						message: e instanceof Error ? e.message : String( e ),
						prop: this.pg_inserts[ 0 ]
					} )
				}
				this.pg_inserts.shift()
			}
			this.replay_timer = undefined
		}

		protected searches = [] as $gd_kit_object_search[]

		search_make( config: Partial<Omit<
			InstanceType<typeof $gd_kit_object_search>,
			'url_base' | 'pipe' | 'destructor'
		>> ) {
			const search: $gd_kit_object_search = this.$.$gd_kit_object_search.make( {
				tenant_id: () => this.tenant_id(),
				...config,
				url_base: () => this.url_read_base(),
				pipe: (item) => this.pipe(item),
				cursor_where: (id, before) => this.cursor_where(id, before),
				compare: (a, b) => this.compare(a, b),
				destructor: () => this.search_destruct(search),
			})

			this.searches.push( search )

			if( !this.replay_timer ) {
				this.replay_timer = new this.$.$mol_after_timeout(
					this.replay_delay(),
					() => $mol_wire_async( this ).replay()
				)
			}

			return search
		}

		private compare( a_id: string, b_id: string ) {
			const [ a, b ] = $mol_wire_race(
				() => this.prop( a_id )?.created_at(),
				() => this.prop( b_id )?.created_at()
			)

			return a > b ? 1 : ( a < b ? -1 : 0 )
		}

		@ $mol_action
		private cursor_where( id: string | null, before = false ) {

			if( !id ) return {}

			const prop = this.prop_data( id )
			return {
				at: { [ before ? 'lt' : 'gt' ]: prop?.at }
			}
		}

		destructor() {
			this.searches = []
			this.replay_timer?.destructor()
		}

		protected search_destruct( search: $gd_kit_object_search ) {
			this.searches = this.searches.filter( src => src !== search )
		}

		protected pipe(item: typeof $gd_kit_prop_data.Value) {
			return this.cache(this.$.$gd_kit_object.ids_globalize(item))
		}

		protected cache(item: typeof $gd_kit_prop_data.Value) {
			if (! item.p) throw new Error('Can\'t cache: no id in item', { cause: { item } })
			this.batch_by_p().cache(item.p, item) // cache props from search

			return item.p
		}

		protected url_read_base() {
			return this.init().url( this.data_type() === 'big' ? 'service_db_wbd' : 'service_db_wmd' )
		}

		protected url_write_base() {
			return this.init().url( this.data_type() === 'big' ? 'service_db_wbd' : 'service_db_wmd' )
		}

		data_type() {
			return 'meta' as 'meta' | 'big'
		}

		protected tenant_id(): string | null {
			return this.init().tenant_id() ?? null
		}

		@ $mol_mem
		lang_inheritance() {
			// todo: use init
			return [ $gd_kit_id.lang_ru ]
		}

		protected normalize_prop( id: string, next: Partial<typeof $gd_kit_prop_data.Value> ) {
			return {
				vt: null,
				vr: null,
				vf: null,
				pt: $gd_kit_id.type_object,
				...next,
				t: next?.t ?? this.tenant_id() ?? undefined,
				p: next?.p ?? id,
				g: next?.g ?? id,
				o: next?.o ?? id,
				s: next?.s ?? id,
			}
		}

		auth_disabled() { return false }

		@ $mol_action
		protected fetch( query: $gd_kit_postgrest_query_type ) {
			const select = Object.keys( $gd_kit_prop_data_list.config.config ).filter( field => field !== 'draft' )

			const params = this.$.$gd_kit_postgrest_query.serialize( { ...query, select } )

			const query_url = `${ this.url_read_base() }prop_get?${ params }`
			const raw = this.$.$gd_kit_transport.get( query_url, {
				auth_token: this.auth_disabled() ? null : undefined
			} ).json()

			return $gd_kit_prop_data_list( raw as any )
		}

		protected split_ids( ids: readonly string[] ) {
			const tenant_default = this.tenant_id() ?? ''
			const tenant_id_groups = [] as { t: string, ids: string[] }[]
			const id_to_typed_id = {} as Record<string, string>
			const chunk_max = 500

			for( let i = 0; i < ids.length; i++ ) {
				const typed_id = ids[ i ]
				const id = $gd_kit_object.id_id( typed_id )

				if( id !== typed_id ) id_to_typed_id[ id ] = typed_id

				const tenant_id = $gd_kit_object.is_id_type_global( typed_id ) ? '' : tenant_default
				let chunk = undefined as undefined | typeof tenant_id_groups[ number ]

				if( chunk?.t !== tenant_id ) {
					chunk = tenant_id_groups.find( rec => rec.t === tenant_id && rec.ids.length < chunk_max )
				}

				if( !chunk || chunk.ids.length >= chunk_max ) {
					chunk = { t: tenant_id, ids: [] }
					tenant_id_groups.push( chunk )
				}

				chunk.ids.push( id )
			}

			return { tenant_id_groups, id_to_typed_id }

		}

		@ $mol_action
		protected batch_by_p_load( ids: readonly string[] ) {
			const { tenant_id_groups, id_to_typed_id } = this.split_ids( ids )

			const data_chunks = $mol_wire_race(
				...tenant_id_groups.map( ( { t, ids } ) =>
					() => this.fetch( { where: { t: t || undefined, p: ids } } )
				)
			)

			const acc = {} as Record<string, typeof $gd_kit_prop_data.Value>

			for (const chunks of data_chunks) {
				for (const item of chunks) {
					const p = id_to_typed_id[item.p] ?? item.p

					acc[p] = this.$.$gd_kit_object.ids_globalize(item, { p })
				}
			}

			return acc
		}

		batch_by_p_patch(
			patches: Record<string, Partial<typeof $gd_kit_prop_data.Value> | null>
		) {
			const upsert_data = [] as Partial<typeof $gd_kit_prop_data.Value>[]
			const delete_object_ids = [] as string[]
			const delete_prop_ids = [] as string[]

			let first = undefined as undefined | string[]

			for( const id in patches ) {
				let patch = patches[ id ]

				if( patch ) {
					patch = {
						...this.$.$gd_kit_object.ids_strip( patch ),

						at: undefined,
						gr: undefined,
						u: undefined,
						changed: undefined,
						deleted: undefined,
						is_git: undefined,
						m2: undefined,
						vj: patch.vj ?? null,
					}

					const keys = Object.keys( patch )
						.filter( k => patch?.[ k as keyof typeof patch ] !== undefined )
						.sort()

					if( !first ) first = keys
					else if( !$mol_compare_deep( first, keys ) ) {
						throw new Error('Keys not match: only current', { cause: {
							patch,
							keys: keys.filter( k => !first?.includes( k ) ),
							first: first.filter( k => !keys?.includes( k ) ),
						} })
					}
				}

				if( patch ) upsert_data.push( patch )
				if( patch !== null ) continue

				if( this.prop( id ).prop_type()?.id() === $gd_kit_id.type_object ) delete_object_ids.push( id )
				else delete_prop_ids.push( id )
			}

			const [ upserted, deleted ] = $mol_wire_race(
				() => upsert_data.length > 0 ? this.upsert( upsert_data ) : undefined,

				() => delete_prop_ids.length > 0 || delete_object_ids.length > 0 ? this.delete( {
					object_ids: delete_object_ids,
					prop_ids: delete_prop_ids,
				} ) : undefined,
			)

			return { ...patches, ...upserted } as NonNullable<typeof upserted>
		}

		protected delete( data: { object_ids: string[], prop_ids: string[] } ) {
			this.$.$gd_kit_transport.post(
				`${ this.url_write_base() }rpc/delete`, {
				body_object: {
					objects: data.object_ids,
					properties: data.prop_ids,
				}
			} )

			return {} as Record<string, null>
		}

		protected upsert( upsert_data: Partial<typeof $gd_kit_prop_data.Value>[] ) {
			const raw = this.$.$gd_kit_transport.post( `${ this.url_write_base() }rpc/upsert`, {
				body_object: upsert_data
			} ).json()

			const rows = $gd_kit_prop_data_list( raw as any )

			return rows.reduce( ( acc, prop ) => {
				acc[ prop.p ] = prop
				return acc
			}, {} as Record<string, typeof $gd_kit_prop_data.Value> )
		}

		private _cache = {} as Record<string, typeof $gd_kit_prop_data.Value | undefined | null>

		protected batch_by_p_cache_dto( id: string, next?: typeof $gd_kit_prop_data.Value | null ) {
			if( next !== undefined ) this._cache[ id ] = next
			return this._cache[ id ]
		}

		@ $mol_mem
		protected batch_by_p() {
			const batch = new this.$.$gd_kit_batch<typeof $gd_kit_prop_data.Value>()
			batch.load = this.batch_by_p_load.bind( this )
			batch.patch = this.batch_by_p_patch.bind( this )
			batch.cache_dto = this.batch_by_p_cache_dto.bind( this )
			return batch
		}

		// Если врубить мемоизацию, отвалится реактивность при обновлении data из промиса
		// Это известная бага в атомах - https://page.hyoo.ru/#!=9sbn4b_lozyyf
		// > В атом сначала протолкнули новое значение, а только потом стали из него вытягивать
		// > то при наивной реализации получается, что подписки сформированы не будут.
		// @ $mol_mem_key
		protected prop_data(
			id: string,
			next?: Partial<typeof $gd_kit_prop_data.Value> | null | false
		): typeof $gd_kit_prop_data.Value | null {
			return this.batch_by_p().data( id, next ? this.normalize_prop( id, next ) : next )
		}

		@ $mol_mem_key
		prop( id: string ) {
			if( !id ) throw new Error(`No id for ${ this.data_type() }`, { cause: { id, type: this.data_type() } } )

			return this.$.$gd_kit_object.make( {
				id: $mol_const( id ),
				registry: () => this,
				data: next => this.prop_data( id, next )
			} )
		}

		object_remove( id: string ) {
			return this.prop_data( id, null )
		}

		@ $mol_action
		create( next: Partial<typeof $gd_kit_prop_data.Value> ) {
			const id = next?.p ?? this.$.$gd_kit_entity.create_id( `${ next?.pt ?? '' }@ ${ next?.vr ?? '' }` )
			const prop = this.prop( id )

			prop.data( this.normalize_prop( id, next ) )
			prop.kids( [] )

			return prop
		}

		@ $mol_action
		protected batch_by_o_pt_load( ids: readonly string[] ) {
			const { tenant_id_groups, id_to_typed_id } = this.split_ids( ids )

			const data_chunks = $mol_wire_race(
				...tenant_id_groups.map( ( { t, ids } ) =>
					() => this.fetch( {
						where: {
							t: t || undefined,
							o: ids,
						}
					} )
				)
			)

			const props_arr = [] as typeof $gd_kit_prop_data.Value[]

			for( const chunks of data_chunks ) {
				for( const item of chunks ) {
					if( item.pt !== $gd_kit_id.type_object ) props_arr.push( item )
				}
			}

			const sorted = props_arr
			sorted.sort( ( ar, br ) => {
				const a = ar.vf ?? 0
				const b = br.vf ?? 0
				return a > b ? 1 : ( a < b ? -1 : 0 )
			} )

			return sorted.reduce( ( acc, raw ) => {
				const owner_id = id_to_typed_id[ raw.o ] ?? raw.o

				const item = this.$.$gd_kit_object.ids_globalize(raw, { p: id_to_typed_id[raw.p] ?? raw.p })

				this.cache(item)

				if( !acc[ owner_id ] ) acc[ owner_id ] = []
				acc[ owner_id ].push( item.p )

				return acc
			}, {} as Record<string, string[]> )
		}

		@ $mol_mem
		protected batch_by_o_pt() {
			const batch = new this.$.$gd_kit_batch<readonly string[], readonly string[]>()
			batch.load = this.batch_by_o_pt_load.bind( this )

			return batch
		}

		prop_kids( owner_id: string ) {
			return this.batch_by_o_pt().data( owner_id )
		}
	}

}