Skip to content

chore(release): staging to production - 2026.03.16 #1026

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
maxtechera merged 2 commits into from
Mar 16, 2026
Merged

chore(release): staging to production - 2026.03.16 #1026

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
d21bca0
feat(AAIDomains): streaming save pipeline — process and save per page…
maxtechera Mar 16, 2026
38cb2a3
Merge pull request #1025 from the-answerai/fix/kumello-68-aaIdomains-oom
maxtechera Mar 16, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
134 changes: 100 additions & 34 deletions packages/components/nodes/documentloaders/AAIDomains/AAIDomains.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -285,6 +285,76 @@ class AAIDomains_DocumentLoaders implements INode {

return docs
}

async loadStream(
nodeData: INodeData,
_: string,
__: ICommonObject,
onBatch: (docs: IDocument[]) => Promise<void>
): Promise<void> {
const textSplitter = nodeData.inputs?.textSplitter as TextSplitter
const limit = nodeData.inputs?.limit ? Number(nodeData.inputs.limit) : undefined
const searchTerm = nodeData.inputs?.searchTerm as string
const includeTags = nodeData.inputs?.includeTags as string
const includeTagsLogic = (nodeData.inputs?.includeTagsLogic as string) || 'OR'
const excludeTags = nodeData.inputs?.excludeTags as string
const isValid = (nodeData.inputs?.isValid as string) || 'all'
const hasAnalysis = (nodeData.inputs?.hasAnalysis as string) || 'all'
const contentFields = nodeData.inputs?.contentFields as string
const metadata = nodeData.inputs?.metadata
const _omitMetadataKeys = nodeData.inputs?.omitMetadataKeys as string

let omitMetadataKeys: string[] = []
if (_omitMetadataKeys) {
omitMetadataKeys = _omitMetadataKeys.split(',').map((key) => key.trim())
}

let parsedContentFields: string[] | null = null
if (contentFields && contentFields.trim()) {
parsedContentFields = contentFields
.split(',')
.map((f) => f.trim())
.filter((f) => f.length > 0)
}

const supabaseUrl = process.env.AAI_DATASTORE_SUPABASE_URL
const supabaseKey = process.env.AAI_DATASTORE_SUPABASE_SERVICE_ROLE_KEY

if (!supabaseUrl || !supabaseKey) {
throw new Error(
'AAI Datastore configuration missing. Please set AAI_DATASTORE_SUPABASE_URL and AAI_DATASTORE_SUPABASE_SERVICE_ROLE_KEY environment variables.'
)
}

const isValidFilter = isValid === 'all' ? null : isValid
const parsedMetadata = metadata ? (typeof metadata === 'object' ? metadata : JSON.parse(metadata)) : null

const loaderOptions: AAIDomainsLoaderParams = {
supabaseUrl,
supabaseKey,
limit: limit || 100,
searchTerm: searchTerm || null,
includeTags: includeTags ? includeTags.split(',').map((t) => t.trim()) : [],
includeTagsLogic,
excludeTags: excludeTags ? excludeTags.split(',').map((t) => t.trim()) : [],
isValid: isValidFilter,
hasAnalysis,
contentFields: parsedContentFields
}

const loader = new AAIDomainsLoader({ ...loaderOptions, textSplitter: textSplitter || null })

await loader.loadStream(async (pageDocs) => {
for (const doc of pageDocs) {
if (_omitMetadataKeys === '*') {
doc.metadata = parsedMetadata ? { ...parsedMetadata } : {}
} else {
doc.metadata = omit({ ...doc.metadata, ...(parsedMetadata || {}) }, omitMetadataKeys)
}
}
await onBatch(pageDocs)
})
}
}

interface AAIDomainsLoaderParams {
Expand Down Expand Up @@ -329,25 +399,24 @@ class AAIDomainsLoader extends BaseDocumentLoader {
this.textSplitter = params.textSplitter ?? null
}

public async load(): Promise<IDocument[]> {
const supabase: SupabaseClient = createClient(this.supabaseUrl, this.supabaseKey, {
private _createSupabaseClient(): SupabaseClient {
return createClient(this.supabaseUrl, this.supabaseKey, {
global: {
fetch: (...args) => {
const [resource, config] = args
// Set 60 second timeout for RPC calls
const controller = new AbortController()
const timeoutId = setTimeout(() => controller.abort(), 60000)

return fetch(resource, {
...config,
signal: controller.signal
}).finally(() => clearTimeout(timeoutId))
return fetch(resource, { ...config, signal: controller.signal }).finally(() => clearTimeout(timeoutId))
}
}
})
}

private async _executeLoad(
supabase: SupabaseClient,
onPage: (pageDocs: IDocument[]) => Promise<void>
): Promise<{ fetchedDomainCount: number }> {
const pageSize = Math.min(this.limit, 100)
let allDocs: IDocument[] = []
let fetchedDomainCount = 0
let lastId: string | null = null
let pageNum = 0
Expand All @@ -368,37 +437,30 @@ class AAIDomainsLoader extends BaseDocumentLoader {

console.info(`[AAIDomains] Fetching page ${pageNum}, size ${currentPageSize}`)

// Retry logic with exponential backoff
let retryCount = 0
const maxRetries = 3
let lastError: any = null
let shouldStopPagination = false

while (retryCount < maxRetries) {
try {
// Build query - dynamic based on contentFields
let selectFields: string

if (this.contentFields && this.contentFields.length > 0) {
// User specified fields - build dynamic select
const baseFields = new Set(['id', 'domain_name', 'created_at', 'updated_at'])
const requestedFields = new Set<string>()

// Parse contentFields to determine which top-level fields we need
for (const field of this.contentFields) {
const topLevelField = field.split('.')[0]
requestedFields.add(topLevelField)
}

// Always include base fields + requested fields
const fieldsToFetch = [...baseFields, ...requestedFields]

// Add tags if not already included
if (!fieldsToFetch.includes('domain_tags')) {
fieldsToFetch.push('domain_tags')
}

// Build select string with tags relation
const fieldsList = fieldsToFetch.filter((f) => f !== 'domain_tags').join(',\n ')
selectFields = `
${fieldsList},
Expand All @@ -408,18 +470,14 @@ class AAIDomainsLoader extends BaseDocumentLoader {
)
`

// Warn about heavy fields
const heavyFields = ['ai_analysis_history', 'ai_analysis_overrides', 'override_metadata', 'source_metadata']
const requestedHeavy = heavyFields.filter((f) => requestedFields.has(f))
if (requestedHeavy.length > 0) {
console.warn(
`[AAIDomains] Warning: Fetching heavy JSONB fields: ${requestedHeavy.join(
', '
)}. This may impact performance.`
`[AAIDomains] Warning: Fetching heavy JSONB fields: ${requestedHeavy.join(', ')}. This may impact performance.`
)
}
} else {
// Default behavior - only lightweight fields
selectFields = `
id,
domain_name,
Expand Down Expand Up @@ -450,7 +508,6 @@ class AAIDomainsLoader extends BaseDocumentLoader {
query = query.gt('id', lastId)
}

// Apply filters
if (this.searchTerm) {
query = query.or(
`domain_name.ilike.%${this.searchTerm}%,meta_title.ilike.%${this.searchTerm}%,meta_description.ilike.%${this.searchTerm}%`
Expand Down Expand Up @@ -491,13 +548,11 @@ class AAIDomainsLoader extends BaseDocumentLoader {
break
}

// Transform domain_tags array to flat tags array (non-mutating)
const domainsWithTags = (data as any[]).map((domain: any) => ({
...domain,
tags: domain.domain_tags?.map((dt: any) => dt.tags).filter(Boolean) || []
}))

// Apply tag filtering if specified
let filteredDomains: any[] = domainsWithTags
if (this.includeTags.length > 0 || this.excludeTags.length > 0) {
filteredDomains = this.filterByTags(domainsWithTags)
Expand All @@ -507,14 +562,13 @@ class AAIDomainsLoader extends BaseDocumentLoader {

const pageDocs = filteredDomains.map((d: any) => this.createDocumentFromDomain(d))

if (this.textSplitter) {
const pageChunks = await this.textSplitter.splitDocuments(pageDocs)
allDocs.push(...pageChunks)
} else {
allDocs.push(...pageDocs)
}
const pageOutput = this.textSplitter
? await this.textSplitter.splitDocuments(pageDocs)
: pageDocs

// Advance cursor only after docs are successfully pushed
await onPage(pageOutput)

// Advance cursor only after onPage callback succeeds
lastId = (data as any[])[data.length - 1].id
pageNum++

Expand Down Expand Up @@ -542,20 +596,32 @@ class AAIDomainsLoader extends BaseDocumentLoader {
}
}

// If we got no data, hit the limit, or all retries failed, break out of pagination loop
if (shouldStopPagination || retryCount >= maxRetries) {
break
}

// Add small delay between successful pages to avoid overwhelming the server
await new Promise((resolve) => setTimeout(resolve, 200))
}

console.info(`[AAIDomains] Load complete. Total documents: ${allDocs.length}, source domains fetched: ${fetchedDomainCount}`)
return { fetchedDomainCount }
}

public async load(): Promise<IDocument[]> {
const supabase = this._createSupabaseClient()
const allDocs: IDocument[] = []
const { fetchedDomainCount } = await this._executeLoad(supabase, async (pageDocs) => {
allDocs.push(...pageDocs)
})
console.info(`[AAIDomains] Load complete. Total documents: ${allDocs.length}, source domains fetched: ${fetchedDomainCount}`)
return allDocs
}

public async loadStream(onBatch: (docs: IDocument[]) => Promise<void>): Promise<void> {
const supabase = this._createSupabaseClient()
const { fetchedDomainCount } = await this._executeLoad(supabase, onBatch)
console.info(`[AAIDomains] Stream complete. Source domains fetched: ${fetchedDomainCount}`)
}

private filterByTags(domains: any[]): any[] {
return domains.filter((domain) => {
const domainTagSlugs = domain.tags?.map((t: any) => t.slug) || []
Expand Down
93 changes: 71 additions & 22 deletions packages/server/src/services/documentstore/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1205,8 +1205,34 @@ const _saveChunksToStorage = async (
//step 1: restore the full paths, if any
await _normalizeFilePaths(appDataSource, data, entity, orgId)

//step 2: split the file into chunks (call _splitIntoChunks directly, not previewChunks)
const docs = await _splitIntoChunks(appDataSource, componentNodes, data, data.userId, data.organizationId)
//step 2: prepare loader node — streaming if supported, batch otherwise
if (!data.loaderId) return
const nodeInstanceFilePath = componentNodes[data.loaderId].filePath as string
const nodeModule = await import(nodeInstanceFilePath)
const docNodeInstance = new nodeModule.nodeClass()

let splitterInstance = null
if (data.splitterId && data.splitterConfig && Object.keys(data.splitterConfig).length > 0) {
const splitterNodeFilePath = componentNodes[data.splitterId].filePath as string
const splitterModule = await import(splitterNodeFilePath)
const splitterNodeInstance = new splitterModule.nodeClass()
splitterInstance = await splitterNodeInstance.init({ inputs: { ...data.splitterConfig }, id: 'splitter_0' })
}

const loaderNodeData = {
credential: data.credential || data.loaderConfig['FLOWISE_CREDENTIAL_ID'] || undefined,
inputs: { ...data.loaderConfig, textSplitter: splitterInstance },
outputs: { output: 'document' }
}
const loaderOptions: ICommonObject = {
chatflowid: data.storeId ?? uuidv4(),
userId: data.userId,
organizationId: data.organizationId,
appDataSource,
databaseEntities,
logger,
processRaw: true
}

//step 3: remove all files associated with the loader
const existingLoaders = JSON.parse(entity.loaders)
Expand Down Expand Up @@ -1291,12 +1317,38 @@ const _saveChunksToStorage = async (
})
).map((chunk) => chunk.id)

if (docs.length > 0) {
//step 7: save new chunks first (safe delete timing — delete AFTER all saved)
let persistedChunks = 0
let persistedChars = 0
let saveFailed = false
//step 7: save chunks — streaming per page if loader supports it, batch otherwise
let persistedChunks = 0
let persistedChars = 0
let saveFailed = false

if (typeof docNodeInstance.loadStream === 'function') {
let chunkOffset = 0
try {
await docNodeInstance.loadStream(loaderNodeData, '', loaderOptions, async (pageDocs: IDocument[]) => {
for (let i = 0; i < pageDocs.length; i += SAVE_BATCH_SIZE) {
const batch = pageDocs.slice(i, i + SAVE_BATCH_SIZE)
const entities = batch.map((chunk: IDocument, localIndex: number) => ({
docId: newLoaderId,
storeId: data.storeId || '',
id: uuidv4(),
chunkNo: chunkOffset + i + localIndex + 1,
pageContent: sanitizeChunkContent(chunk.pageContent),
metadata: JSON.stringify(chunk.metadata),
userId: data.userId,
organizationId: data.organizationId
}))
await chunkRepository.insert(entities) // insert() skips TypeORM lifecycle hooks — safe: DocumentStoreFileChunk has none
persistedChunks += batch.length
persistedChars += batch.reduce((acc: number, chunk: IDocument) => acc + (chunk.pageContent?.length ?? 0), 0)
}
chunkOffset += pageDocs.length
})
} catch (streamError) {
saveFailed = true
}
} else {
const docs: IDocument[] = await docNodeInstance.init(loaderNodeData, '', loaderOptions)
for (let i = 0; i < docs.length; i += SAVE_BATCH_SIZE) {
const batch = docs.slice(i, i + SAVE_BATCH_SIZE)
try {
Expand All @@ -1321,25 +1373,22 @@ const _saveChunksToStorage = async (
break
}
}
}

if (!saveFailed) {
//step 8: delete old chunks only after all new chunks are confirmed saved
if (existingChunkIds.length > 0) {
for (let i = 0; i < existingChunkIds.length; i += DELETE_BATCH_SIZE) {
await chunkRepository.delete({ id: In(existingChunkIds.slice(i, i + DELETE_BATCH_SIZE)) })
}
//step 8: delete old chunks only after all new chunks are confirmed saved
if (!saveFailed) {
if (existingChunkIds.length > 0) {
for (let i = 0; i < existingChunkIds.length; i += DELETE_BATCH_SIZE) {
await chunkRepository.delete({ id: In(existingChunkIds.slice(i, i + DELETE_BATCH_SIZE)) })
}
loader.totalChunks = persistedChunks
loader.totalChars = persistedChars
loader.status = 'SYNC'
} else {
// Partial failure: record actual persisted count, mark STALE
loader.totalChunks = persistedChunks
loader.totalChars = persistedChars
loader.status = DocumentStoreStatus.STALE
}
} else {
loader.totalChunks = persistedChunks
loader.totalChars = persistedChars
loader.status = 'SYNC'
} else {
loader.totalChunks = persistedChunks
loader.totalChars = persistedChars
loader.status = DocumentStoreStatus.STALE
}
// have a flag and iterate over the loaders and update the entity status to SYNC
const allSynced = existingLoaders.every((ldr: IDocumentStoreLoader) => ldr.status === 'SYNC')
Expand Down
Loading