diff --git a/packages/core/src/db/schemas.ts b/packages/core/src/db/schemas.ts index edcaa015..09b33e2f 100644 --- a/packages/core/src/db/schemas.ts +++ b/packages/core/src/db/schemas.ts @@ -370,6 +370,12 @@ export const UserDataSchema = z.object({ // }) // ) // .optional(), + dynamicAddonFetching: z + .object({ + enabled: z.boolean().optional(), + condition: z.string().min(1).max(200).optional(), + }) + .optional(), groups: z .object({ enabled: z.boolean().optional(), diff --git a/packages/core/src/parser/streamExpression.ts b/packages/core/src/parser/streamExpression.ts index 64fbdb33..7229d5cf 100644 --- a/packages/core/src/parser/streamExpression.ts +++ b/packages/core/src/parser/streamExpression.ts @@ -607,6 +607,26 @@ export abstract class StreamExpressionEngine { } } +export class ExitConditionEvaluator extends StreamExpressionEngine { + constructor( + private totalStreams: ParsedStream[], + private totalTimeTaken: number + ) { + super(); + this.parser.consts.totalStreams = this.totalStreams; + this.parser.consts.totalTimeTaken = this.totalTimeTaken; + } + + async evaluate(condition: string) { + return await this.evaluateCondition(condition); + } + + static async testEvaluate(condition: string) { + const parser = new ExitConditionEvaluator([], 0); + return await parser.evaluate(condition); + } +} + export class GroupConditionEvaluator extends StreamExpressionEngine { private previousStreams: ParsedStream[]; private totalStreams: ParsedStream[]; diff --git a/packages/core/src/streams/fetcher.ts b/packages/core/src/streams/fetcher.ts index 6bf2eec5..01fdaa98 100644 --- a/packages/core/src/streams/fetcher.ts +++ b/packages/core/src/streams/fetcher.ts @@ -7,7 +7,10 @@ import { getTimeTakenSincePoint, } from '../utils/index.js'; import { Wrapper } from '../wrapper.js'; -import { GroupConditionEvaluator } from '../parser/streamExpression.js'; +import { + ExitConditionEvaluator, + GroupConditionEvaluator, +} from '../parser/streamExpression.js'; import StreamFilter from './filterer.js'; import StreamPrecompute from './precomputer.js'; import StreamDeduplicator from './deduplicator.js'; @@ -135,7 +138,7 @@ class StreamFetcher { (s) => s.type !== constants.ERROR_STREAM_TYPE ), errors: addonErrors, - statistics: statisticStream, + statistic: statisticStream, timeTaken: Date.now() - start, }; } catch (error) { @@ -155,7 +158,6 @@ class StreamFetcher { return { success: false as const, errors: [addonErrors], - statistics: [], timeTaken: 0, streams: [], }; @@ -171,7 +173,9 @@ class StreamFetcher { const groupStreams = results.flatMap((r) => r.streams); const groupErrors = results.flatMap((r) => r.errors); - const groupStatistics = results.flatMap((r) => r.statistics); + const groupStatistics = results + .flatMap((r) => r.statistic) + .filter((s) => s !== undefined); const filteredStreams = await this.deduplicate.deduplicate( await this.filter.filter(groupStreams, type, id) @@ -190,7 +194,62 @@ class StreamFetcher { }; // If groups are configured, handle group-based fetching - if ( + if (this.userData.dynamicAddonFetching?.enabled) { + const condition = this.userData.dynamicAddonFetching.condition; + if (!condition) { + throw new Error('Dynamic addon fetching condition is not set'); + } + + await new Promise((resolve) => { + let activePromises = addons.length; + if (activePromises === 0) { + resolve(); + return; + } + + const checkExit = async () => { + const timeTaken = Date.now() - start; + const evaluator = new ExitConditionEvaluator(allStreams, timeTaken); + const shouldExit = await evaluator.evaluate(condition); + + if (shouldExit) { + logger.info( + `Exit condition met with results from ${addons.length - activePromises} addons. (${activePromises} addons still fetching) Returning results.` + ); + resolve(); + } + }; + + addons.forEach((addon) => { + fetchFromAddon(addon) + .then(async (result) => { + allStreams.push(...result.streams); + allErrors.push(...result.errors); + if (result.statistic) { + allStatisticStreams.push(result.statistic); + } + await checkExit(); + }) + .catch((error) => { + logger.error( + `Unhandled error from fetchFromAddon for ${getAddonName(addon)}:`, + error + ); + allErrors.push({ + title: `[❌] ${getAddonName(addon)}`, + description: + error instanceof Error ? error.message : String(error), + }); + }) + .finally(() => { + activePromises--; + if (activePromises === 0) { + resolve(); + } + }); + }); + }); + } else if ( this.userData.groups?.groupings && this.userData.groups.groupings.length > 0 && this.userData.groups.enabled !== false @@ -219,6 +278,7 @@ class StreamFetcher { const groupAddons = addons.filter( (addon) => addon.preset.id && group.addons.includes(addon.preset.id) ); + if (groupAddons.length === 0) return Promise.resolve(null); logger.info( `Queueing parallel fetch for group with ${groupAddons.length} addons.` ); @@ -226,48 +286,52 @@ class StreamFetcher { }); for (let i = 0; i < this.userData.groups.groupings.length; i++) { - const groupResult = await groupPromises[i]; - const group = this.userData.groups.groupings[i]; + const groupPromise = groupPromises[i]; if (i === 0) { + const groupResult = await groupPromise; + if (!groupResult) continue; allStreams.push(...groupResult.streams); allErrors.push(...groupResult.errors); allStatisticStreams.push(...groupResult.statistics); totalTimeTaken = groupResult.totalTime; previousGroupStreams = groupResult.streams; previousGroupTimeTaken = groupResult.totalTime; + continue; + } + // For groups other than the first, check their condition + const group = this.userData.groups.groupings[i]; + if (!group.condition || !group.addons.length) continue; + + const evaluator = new GroupConditionEvaluator( + previousGroupStreams, + allStreams, + previousGroupTimeTaken, + totalTimeTaken, + queryType + ); + const shouldIncludeAndContinue = await evaluator.evaluate( + group.condition + ); + + if (shouldIncludeAndContinue) { + logger.info( + `Condition met for parallel group ${i + 1}, awaiting its streams and continuing.` + ); + const groupResult = await groupPromise; + if (!groupResult) continue; + allStreams.push(...groupResult.streams); + allErrors.push(...groupResult.errors); + allStatisticStreams.push(...groupResult.statistics); + totalTimeTaken = Math.max(totalTimeTaken, groupResult.totalTime); + previousGroupStreams = groupResult.streams; + previousGroupTimeTaken = groupResult.totalTime; } else { - // For groups other than the first, check their condition - if (!group.condition || !group.addons.length) continue; - - const evaluator = new GroupConditionEvaluator( - previousGroupStreams, - allStreams, - previousGroupTimeTaken, - totalTimeTaken, - queryType + logger.info( + `Condition not met for parallel group ${i + 1}, skipping remaining groups.` ); - const shouldIncludeAndContinue = await evaluator.evaluate( - group.condition - ); - - if (shouldIncludeAndContinue) { - logger.info( - `Condition met for parallel group ${i + 1}, including streams and continuing.` - ); - allStreams.push(...groupResult.streams); - allErrors.push(...groupResult.errors); - allStatisticStreams.push(...groupResult.statistics); - totalTimeTaken = Math.max(totalTimeTaken, groupResult.totalTime); - previousGroupStreams = groupResult.streams; - previousGroupTimeTaken = groupResult.totalTime; - } else { - logger.info( - `Condition not met for parallel group ${i + 1}, skipping remaining groups.` - ); - // exit early. - break; - } + // exit early. + break; } } } else { diff --git a/packages/core/src/utils/config.ts b/packages/core/src/utils/config.ts index 23dce79d..4269971c 100644 --- a/packages/core/src/utils/config.ts +++ b/packages/core/src/utils/config.ts @@ -24,6 +24,7 @@ import { } from './index.js'; import { ZodError } from 'zod'; import { + ExitConditionEvaluator, GroupConditionEvaluator, StreamSelector, } from '../parser/streamExpression.js'; @@ -364,6 +365,19 @@ export async function validateConfig( } } + if ( + config.dynamicAddonFetching?.condition && + config.dynamicAddonFetching.enabled + ) { + try { + await ExitConditionEvaluator.testEvaluate( + config.dynamicAddonFetching.condition + ); + } catch (error) { + throw new Error(`Invalid dynamic addon fetching condition: ${error}`); + } + } + // validate excluded filter condition const streamExpressions = [ ...(config.excludedStreamExpressions ?? []), diff --git a/packages/frontend/src/components/menu/addons.tsx b/packages/frontend/src/components/menu/addons.tsx index a99ddd0b..d1ed3dfa 100644 --- a/packages/frontend/src/components/menu/addons.tsx +++ b/packages/frontend/src/components/menu/addons.tsx @@ -429,7 +429,7 @@ function Content() { {userData.presets.length > 0 && } {userData.presets.length > 0 && mode === 'pro' && ( - + )} )} @@ -1186,8 +1186,13 @@ function AddonFilterPopover({ ); } -function AddonGroupCard() { +function AddonFetchingBehaviorCard() { const { userData, setUserData } = useUserData(); + const [mode, setMode] = useState(() => { + if (userData.dynamicAddonFetching?.enabled) return 'dynamic'; + if (userData.groups?.enabled) return 'groups'; + return 'default'; + }); // Helper function to get presets that are not in any group except the current one const getAvailablePresets = (currentGroupIndex: number) => { @@ -1213,23 +1218,15 @@ function AddonGroupCard() { updates: Partial<{ addons: string[]; condition: string }> ) => { setUserData((prev) => { - // Initialize groups array if it doesn't exist const currentGroups = prev.groups?.groupings || []; - - // Create a new array with all existing groups const newGroups = [...currentGroups]; - - // Update the specific group with new values, preserving other fields newGroups[index] = { ...newGroups[index], ...updates, }; - if (index === 0) { - // set condition for first group to true newGroups[index].condition = 'true'; } - return { ...prev, groups: { @@ -1240,131 +1237,164 @@ function AddonGroupCard() { }); }; + const handleModeChange = (newMode: string) => { + setMode(newMode); + setUserData((prev) => ({ + ...prev, + groups: { + ...prev.groups, + enabled: newMode === 'groups', + }, + dynamicAddonFetching: { + ...prev.dynamicAddonFetching, + enabled: newMode === 'dynamic', + }, + })); + }; + + const descriptions = { + default: + 'Fetch from all addons simultaneously and wait for all addons to finish fetching before returning results.', + groups: + 'Organize addons into groups. Streams are fetched based on group conditions, either sequentially or in parallel.', + dynamic: + 'Fetch from all addons at once, and exit once a specified condition is met.', + }; + return ( -
- Optionally assign your addons to groups. Streams are only fetched from - your first group initially, and only if a certain condition is met, will - streams be fetched from the next group, and so on. Leaving this blank - will mean streams are fetched from all addons. Check the{' '} - - wiki - {' '} - for a detailed guide to using groups. -
- { - setUserData((prev) => ({ - ...prev, - groups: { ...prev.groups, enabled: value }, - })); - }} - side="right" - /> { + setUserData((prev) => ({ ...prev, groups: { ...prev.groups, - groupings: [...currentGroups, { addons: [], condition: '' }], + behaviour: value as 'sequential' | 'parallel', }, - }; - }); + })); + }} + options={[ + { label: 'Parallel', value: 'parallel' }, + { label: 'Sequential', value: 'sequential' }, + ]} + help={ + userData.groups?.behaviour === 'sequential' + ? 'Streams are fetched from the first group only to begin with. If the condition for the next group is met, streams are fetched from the next group, and so on.' + : 'Streams are fetched from all groups at the same time. When a condition is not met, results from its group onwards are simply not shown.' + } + /> + + {(userData.groups?.groupings || []).map((group, index) => ( +
+
+
+ { + updateGroup(index, { addons: value }); + }} + /> +
+
+ { + updateGroup(index, { condition: value }); + }} + /> +
+
+ } + intent="alert-subtle" + onClick={() => { + setUserData((prev) => { + const newGroups = [...(prev.groups?.groupings || [])]; + newGroups.splice(index, 1); + return { + ...prev, + groups: { ...prev.groups, groupings: newGroups }, + }; + }); + }} + /> +
+ ))} + +
+ } + onClick={() => { + setUserData((prev) => { + const currentGroups = prev.groups?.groupings || []; + return { + ...prev, + groups: { + ...prev.groups, + groupings: [ + ...currentGroups, + { addons: [], condition: '' }, + ], + }, + }; + }); + }} + /> +
+ + )} + + {mode === 'dynamic' && ( + { + setUserData((prev) => ({ + ...prev, + dynamicAddonFetching: { + ...prev.dynamicAddonFetching, + condition: value, + }, + })); }} + help="When this condition is met, no more addons will be fetched from" /> - + )} ); }