From 59a927aec3121f8547e4a355c6dcb85fcb28a03e Mon Sep 17 00:00:00 2001 From: Riley Des Date: Tue, 2 Jun 2026 11:44:04 -0500 Subject: [PATCH 1/4] feat(vectorstores): add Valkey vector store integration - Add ValkeyApi credential for host/port/auth configuration - Add ValkeyUrlApi credential for connection URL-based setup - Implement ValkeyVectorStore node with full CRUD operations - Support vector similarity search with metadata filtering - Add comprehensive integration tests with Valkey 9.1+ and valkey-search v1.2+ module - Add unit tests covering core functionality and edge cases - Include Valkey icon asset for UI representation - Update package.json with valkey-glide dependency Signed-off-by: Riley Des --- .../credentials/ValkeyApi.credential.ts | 52 ++ .../credentials/ValkeyUrlApi.credential.ts | 26 + .../vectorstores/Valkey/Valkey.int.test.ts | 482 +++++++++++ .../nodes/vectorstores/Valkey/Valkey.test.ts | 798 ++++++++++++++++++ .../nodes/vectorstores/Valkey/Valkey.ts | 698 +++++++++++++++ .../nodes/vectorstores/Valkey/valkey.svg | 1 + packages/components/package.json | 1 + pnpm-lock.yaml | 73 +- 8 files changed, 2130 insertions(+), 1 deletion(-) create mode 100644 packages/components/credentials/ValkeyApi.credential.ts create mode 100644 packages/components/credentials/ValkeyUrlApi.credential.ts create mode 100644 packages/components/nodes/vectorstores/Valkey/Valkey.int.test.ts create mode 100644 packages/components/nodes/vectorstores/Valkey/Valkey.test.ts create mode 100644 packages/components/nodes/vectorstores/Valkey/Valkey.ts create mode 100644 packages/components/nodes/vectorstores/Valkey/valkey.svg diff --git a/packages/components/credentials/ValkeyApi.credential.ts b/packages/components/credentials/ValkeyApi.credential.ts new file mode 100644 index 00000000000..9887f3396e7 --- /dev/null +++ b/packages/components/credentials/ValkeyApi.credential.ts @@ -0,0 +1,52 @@ +import { INodeParams, INodeCredential } from '../src/Interface' + +class ValkeyApi implements INodeCredential { + label: string + name: string + version: number + description: string + inputs: INodeParams[] + + constructor() { + this.label = 'Valkey API' + this.name = 'valkeyApi' + this.version = 1.0 + this.description = 'Connect to Valkey using host, port, and optional authentication' + this.inputs = [ + { + label: 'Valkey Host', + name: 'valkeyHost', + type: 'string', + default: '127.0.0.1' + }, + { + label: 'Port', + name: 'valkeyPort', + type: 'number', + default: '6379' + }, + { + label: 'User', + name: 'valkeyUser', + type: 'string', + placeholder: '', + optional: true + }, + { + label: 'Password', + name: 'valkeyPassword', + type: 'password', + placeholder: '', + optional: true + }, + { + label: 'Use TLS', + name: 'valkeyTls', + type: 'boolean', + optional: true + } + ] + } +} + +module.exports = { credClass: ValkeyApi } diff --git a/packages/components/credentials/ValkeyUrlApi.credential.ts b/packages/components/credentials/ValkeyUrlApi.credential.ts new file mode 100644 index 00000000000..6ede571c5f7 --- /dev/null +++ b/packages/components/credentials/ValkeyUrlApi.credential.ts @@ -0,0 +1,26 @@ +import { INodeParams, INodeCredential } from '../src/Interface' + +class ValkeyUrlApi implements INodeCredential { + label: string + name: string + version: number + description: string + inputs: INodeParams[] + + constructor() { + this.label = 'Valkey URL' + this.name = 'valkeyUrlApi' + this.version = 1.0 + this.description = 'Connect to Valkey using a connection URL' + this.inputs = [ + { + label: 'Valkey URL', + name: 'valkeyUrl', + type: 'password', + placeholder: 'valkey://user:password@localhost:6379' + } + ] + } +} + +module.exports = { credClass: ValkeyUrlApi } diff --git a/packages/components/nodes/vectorstores/Valkey/Valkey.int.test.ts b/packages/components/nodes/vectorstores/Valkey/Valkey.int.test.ts new file mode 100644 index 00000000000..a9d21a3440a --- /dev/null +++ b/packages/components/nodes/vectorstores/Valkey/Valkey.int.test.ts @@ -0,0 +1,482 @@ +/** + * Integration tests for ValkeyVectorStore. + * + * Requirements: + * - Valkey 9.1+ with valkey-search 1.2+ module loaded + * - Default: localhost:6379 (override with VALKEY_HOST / VALKEY_PORT) + * + * Tests are automatically skipped if Valkey is unavailable. + */ + +jest.mock('../../../src/utils', () => ({ + getBaseClasses: jest.fn(() => ['VectorStore']), + getCredentialData: jest.fn(), + getCredentialParam: jest.fn() +})) + +import { GlideClient } from '@valkey/valkey-glide' +import { Document } from '@langchain/core/documents' + +const { ValkeyVectorStore } = require('./Valkey') + +const VALKEY_HOST = process.env.VALKEY_HOST || '127.0.0.1' +const VALKEY_PORT = parseInt(process.env.VALKEY_PORT || '6379', 10) + +// Deterministic fake embeddings — produces normalized, distinct vectors per text +class FakeEmbeddings { + async embedDocuments(texts: string[]): Promise { + return texts.map((t) => this.embed(t)) + } + async embedQuery(text: string): Promise { + return this.embed(text) + } + private embed(text: string): number[] { + const vec = new Array(128).fill(0) + for (let i = 0; i < text.length; i++) { + vec[i % 128] += text.charCodeAt(i) / 1000 + } + const mag = Math.sqrt(vec.reduce((s: number, v: number) => s + v * v, 0)) || 1 + return vec.map((v: number) => v / mag) + } +} + +async function checkValkeyAvailability(): Promise { + let client: GlideClient | null = null + try { + client = await GlideClient.createClient({ + addresses: [{ host: VALKEY_HOST, port: VALKEY_PORT }], + requestTimeout: 3000 + }) + } catch (err: any) { + return `Cannot connect to Valkey at ${VALKEY_HOST}:${VALKEY_PORT}: ${err.message}` + } + try { + await client.customCommand(['FT._LIST']) + } catch (err: any) { + client.close() + return `valkey-search module not loaded at ${VALKEY_HOST}:${VALKEY_PORT}: ${err.message}` + } + client.close() + return null +} + +describe('ValkeyVectorStore Integration', () => { + let client: GlideClient + let unavailableReason: string | null = null + const testId = Date.now() + const indexName = `test-${testId}` + + beforeAll(async () => { + unavailableReason = await checkValkeyAvailability() + if (unavailableReason) { + console.warn(`\n⚠️ Skipping integration tests: ${unavailableReason}\n`) + return + } + client = await GlideClient.createClient({ + addresses: [{ host: VALKEY_HOST, port: VALKEY_PORT }], + requestTimeout: 5000 + }) + }, 15000) + + afterAll(async () => { + if (!client) return + // Clean up all test indices and keys + try { + const indices = (await client.customCommand(['FT._LIST'])) as string[] + for (const idx of indices) { + if (String(idx).includes(String(testId))) { + await client.customCommand(['FT.DROPINDEX', idx]).catch(() => {}) + } + } + } catch { + /* cleanup best-effort */ + } + try { + const [, keys] = await client.scan('0', { match: `test:${testId}*`, count: 1000 }) + if (keys.length) await client.del(keys as string[]) + } catch { + /* cleanup best-effort */ + } + client.close() + }, 10000) + + function skip() { + return !!unavailableReason + } + + function makeStore(suffix: string, opts: Record = {}) { + return new ValkeyVectorStore(new FakeEmbeddings(), { + valkeyClient: client, + indexName: `${indexName}-${suffix}`, + keyPrefix: `test:${testId}-${suffix}:`, + ...opts + }) + } + + describe('index lifecycle', () => { + it('should return empty results on empty index', async () => { + if (skip()) return + const store = makeStore('empty') + const results = await store.similaritySearchVectorWithScore(await new FakeEmbeddings().embedQuery('x'), 5) + expect(results).toEqual([]) + }) + + it('should create index and upsert documents', async () => { + if (skip()) return + const store = makeStore('lifecycle') + await store.addDocuments([ + new Document({ pageContent: 'Valkey is fast', metadata: { source: 'a' } }), + new Document({ pageContent: 'Vector search is useful', metadata: { source: 'b' } }) + ]) + expect(await store.checkIndexExists()).toBe(true) + }) + + it('should search and return correct documents', async () => { + if (skip()) return + const store = makeStore('lifecycle') + const results = await store.similaritySearchVectorWithScore(await new FakeEmbeddings().embedQuery('Valkey is fast'), 2) + expect(results.length).toBe(2) + expect(results[0][0].pageContent).toBe('Valkey is fast') + expect(results[0][0].metadata.source).toBe('a') + }) + + it('should respect top K', async () => { + if (skip()) return + const store = makeStore('lifecycle') + const results = await store.similaritySearchVectorWithScore(await new FakeEmbeddings().embedQuery('search'), 1) + expect(results.length).toBe(1) + }) + + it('should drop and recreate index', async () => { + if (skip()) return + const store = makeStore('lifecycle') + expect(await store.dropIndex()).toBe(true) + expect(await store.checkIndexExists()).toBe(false) + await store.addDocuments([new Document({ pageContent: 'recreated', metadata: {} })]) + expect(await store.checkIndexExists()).toBe(true) + }) + }) + + describe('deleteAll cleans up HASH keys', () => { + it('should remove both index and all HASH keys', async () => { + if (skip()) return + const prefix = `test:${testId}-delall:` + const store = makeStore('delall') + await store.addDocuments([ + new Document({ pageContent: 'doc1', metadata: {} }), + new Document({ pageContent: 'doc2', metadata: {} }) + ]) + + // Verify keys exist + const [, keysBefore] = await client.scan('0', { match: `${prefix}*`, count: 100 }) + expect(keysBefore.length).toBe(2) + + await store.delete({ deleteAll: true }) + + // Verify index gone + expect(await store.checkIndexExists()).toBe(false) + // Verify HASH keys gone + const [, keysAfter] = await client.scan('0', { match: `${prefix}*`, count: 100 }) + expect(keysAfter.length).toBe(0) + }) + }) + + describe('delete by ids', () => { + it('should delete specific documents by id', async () => { + if (skip()) return + const prefix = `test:${testId}-delid:` + const store = makeStore('delid') + const keys = [`${prefix}a`, `${prefix}b`, `${prefix}c`] + await store.addDocuments( + [ + new Document({ pageContent: 'doc a', metadata: {} }), + new Document({ pageContent: 'doc b', metadata: {} }), + new Document({ pageContent: 'doc c', metadata: {} }) + ], + { keys } + ) + + await store.delete({ ids: [`${prefix}a`, `${prefix}c`] }) + expect(await client.customCommand(['EXISTS', `${prefix}a`])).toBe(0) + expect(await client.customCommand(['EXISTS', `${prefix}b`])).toBe(1) + expect(await client.customCommand(['EXISTS', `${prefix}c`])).toBe(0) + }) + + it('should no-op for empty ids array', async () => { + if (skip()) return + const store = makeStore('delempty') + // Should not throw + await store.delete({ ids: [] }) + }) + + it('should not throw for non-existent ids', async () => { + if (skip()) return + const store = makeStore('delnoexist') + await store.delete({ ids: ['nonexistent-key-xyz'] }) + }) + }) + + describe('auto-generated UUID keys', () => { + it('should generate unique keys and not overwrite on concurrent adds', async () => { + if (skip()) return + const prefix = `test:${testId}-uuid:` + const store = makeStore('uuid') + + // Two parallel adds + await Promise.all([ + store.addDocuments([new Document({ pageContent: 'parallel 1', metadata: {} })]), + store.addDocuments([new Document({ pageContent: 'parallel 2', metadata: {} })]) + ]) + + const [, keys] = await client.scan('0', { match: `${prefix}*`, count: 100 }) + expect(keys.length).toBe(2) + // Keys should be different + expect(keys[0]).not.toBe(keys[1]) + }) + + it('should produce UUID-format keys', async () => { + if (skip()) return + const prefix = `test:${testId}-uuidfmt:` + const store = makeStore('uuidfmt') + await store.addDocuments([new Document({ pageContent: 'test', metadata: {} })]) + + const [, keys] = await client.scan('0', { match: `${prefix}*`, count: 10 }) + expect(keys.length).toBe(1) + const key = String(keys[0]) + const uuidPart = key.replace(prefix, '') + expect(uuidPart).toMatch(/^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/) + }) + }) + + describe('user-provided keys', () => { + it('should store and retrieve with custom keys', async () => { + if (skip()) return + const prefix = `test:${testId}-custom:` + const store = makeStore('custom') + const keys = [`${prefix}first`, `${prefix}second`] + + await store.addDocuments( + [ + new Document({ pageContent: 'custom key doc 1', metadata: { idx: 0 } }), + new Document({ pageContent: 'custom key doc 2', metadata: { idx: 1 } }) + ], + { keys } + ) + + expect(await client.customCommand(['EXISTS', keys[0]])).toBe(1) + expect(await client.customCommand(['EXISTS', keys[1]])).toBe(1) + }) + }) + + describe('batch upsert', () => { + it('should handle more documents than batchSize', async () => { + if (skip()) return + const store = makeStore('batch') + const docs = Array.from({ length: 7 }, (_, i) => new Document({ pageContent: `Batch doc ${i}`, metadata: { i } })) + await store.addDocuments(docs, { batchSize: 3 }) + + const results = await store.similaritySearchVectorWithScore(await new FakeEmbeddings().embedQuery('Batch doc'), 7) + expect(results.length).toBe(7) + }) + }) + + describe('TTL support', () => { + it('should set TTL on all documents', async () => { + if (skip()) return + const prefix = `test:${testId}-ttl:` + const store = makeStore('ttl', { ttl: 120 }) + await store.addDocuments([ + new Document({ pageContent: 'ttl doc 1', metadata: {} }), + new Document({ pageContent: 'ttl doc 2', metadata: {} }) + ]) + + // Collect all keys via scan loop + const allKeys: string[] = [] + let cursor = '0' + do { + const [next, keys] = await client.scan(cursor, { match: `${prefix}*`, count: 100 }) + cursor = String(next) + allKeys.push(...(keys as string[])) + } while (cursor !== '0') + + expect(allKeys.length).toBe(2) + for (const key of allKeys) { + const ttl = await client.ttl(key) + expect(ttl).toBeGreaterThan(0) + expect(ttl).toBeLessThanOrEqual(120) + } + }) + + it('should not set TTL when not configured', async () => { + if (skip()) return + const prefix = `test:${testId}-nottl:` + const store = makeStore('nottl') + await store.addDocuments([new Document({ pageContent: 'no ttl doc', metadata: {} })]) + + const [, keys] = await client.scan('0', { match: `${prefix}*`, count: 10 }) + expect(keys.length).toBe(1) + const ttl = await client.ttl(keys[0] as string) + expect(ttl).toBe(-1) // -1 means no expiry + }) + }) + + describe('metadata filtering', () => { + it('should filter results by metadata content', async () => { + if (skip()) return + const store = makeStore('filter') + await store.addDocuments([ + new Document({ pageContent: 'apple fruit', metadata: { category: 'fruit' } }), + new Document({ pageContent: 'banana fruit', metadata: { category: 'fruit' } }), + new Document({ pageContent: 'carrot vegetable', metadata: { category: 'vegetable' } }) + ]) + + const queryVec = await new FakeEmbeddings().embedQuery('food') + // Filter for 'fruit' — TEXT search within metadata JSON + const results = await store.similaritySearchVectorWithScore(queryVec, 3, ['fruit']) + expect(results.length).toBe(2) + for (const [doc] of results) { + expect(doc.metadata.category).toBe('fruit') + } + }) + + it('should support OR filtering with multiple terms', async () => { + if (skip()) return + const store = makeStore('filter') // reuse index from above + const queryVec = await new FakeEmbeddings().embedQuery('food') + const results = await store.similaritySearchVectorWithScore(queryVec, 3, ['fruit', 'vegetable']) + expect(results.length).toBe(3) + }) + + it('should return no results when filter matches nothing', async () => { + if (skip()) return + const store = makeStore('filter') + const queryVec = await new FakeEmbeddings().embedQuery('food') + const results = await store.similaritySearchVectorWithScore(queryVec, 3, ['nonexistent']) + expect(results.length).toBe(0) + }) + + it('should throw when both instance and method filter provided', async () => { + if (skip()) return + const store = makeStore('filterboth', { filter: ['tag1'] }) + const queryVec = await new FakeEmbeddings().embedQuery('x') + await expect(store.similaritySearchVectorWithScore(queryVec, 1, ['tag2'])).rejects.toThrow('cannot provide both') + }) + + it('should apply constructor filter on search', async () => { + if (skip()) return + const store = new ValkeyVectorStore(new FakeEmbeddings(), { + valkeyClient: client, + indexName: `${indexName}-filter`, // reuse existing index with data + keyPrefix: `test:${testId}-filter:`, + filter: ['vegetable'] + }) + const queryVec = await new FakeEmbeddings().embedQuery('food') + const results = await store.similaritySearchVectorWithScore(queryVec, 3) + expect(results.length).toBe(1) + expect(results[0][0].metadata.category).toBe('vegetable') + }) + }) + + describe('search result ordering', () => { + it('should return most similar document first with ascending scores', async () => { + if (skip()) return + const store = makeStore('order') + await store.addDocuments([ + new Document({ pageContent: 'completely unrelated cooking recipe for pasta', metadata: {} }), + new Document({ pageContent: 'Valkey vector similarity search', metadata: {} }), + new Document({ pageContent: 'random text about weather forecast', metadata: {} }) + ]) + + const queryVec = await new FakeEmbeddings().embedQuery('Valkey vector similarity search') + const results = await store.similaritySearchVectorWithScore(queryVec, 3) + expect(results.length).toBe(3) + expect(results[0][0].pageContent).toContain('Valkey vector') + // Scores ascending (lower = more similar for cosine distance) + for (let i = 0; i < results.length - 1; i++) { + expect(results[i][1]).toBeLessThanOrEqual(results[i + 1][1]) + } + }) + }) + + describe('metadata preservation', () => { + it('should preserve complex metadata through upsert and search', async () => { + if (skip()) return + const store = makeStore('meta') + const meta = { source: 'test', page: 42, tags: ['a', 'b'], nested: { key: 'value' } } + await store.addDocuments([new Document({ pageContent: 'metadata test', metadata: meta })]) + + const results = await store.similaritySearchVectorWithScore(await new FakeEmbeddings().embedQuery('metadata test'), 1) + expect(results.length).toBe(1) + expect(results[0][0].metadata).toEqual(meta) + }) + }) + + describe('static factory methods', () => { + it('fromDocuments should create and populate store', async () => { + if (skip()) return + const store = await ValkeyVectorStore.fromDocuments( + [new Document({ pageContent: 'factory doc', metadata: { via: 'factory' } })], + new FakeEmbeddings(), + { valkeyClient: client, indexName: `${indexName}-factory`, keyPrefix: `test:${testId}-factory:` } + ) + const results = await store.similaritySearchVectorWithScore(await new FakeEmbeddings().embedQuery('factory doc'), 1) + expect(results.length).toBe(1) + expect(results[0][0].metadata.via).toBe('factory') + }) + + it('fromTexts should create store with array metadata', async () => { + if (skip()) return + const store = await ValkeyVectorStore.fromTexts(['text one', 'text two'], [{ idx: 0 }, { idx: 1 }], new FakeEmbeddings(), { + valkeyClient: client, + indexName: `${indexName}-fromtxt`, + keyPrefix: `test:${testId}-fromtxt:` + }) + const results = await store.similaritySearchVectorWithScore(await new FakeEmbeddings().embedQuery('text one'), 2) + expect(results.length).toBe(2) + }) + + it('fromTexts should apply single metadata object to all docs', async () => { + if (skip()) return + const store = await ValkeyVectorStore.fromTexts(['a', 'b'], { shared: true }, new FakeEmbeddings(), { + valkeyClient: client, + indexName: `${indexName}-fromtxt2`, + keyPrefix: `test:${testId}-fromtxt2:` + }) + const results = await store.similaritySearchVectorWithScore(await new FakeEmbeddings().embedQuery('a'), 2) + for (const [doc] of results) { + expect(doc.metadata.shared).toBe(true) + } + }) + }) + + describe('edge cases', () => { + it('should handle documents with empty pageContent', async () => { + if (skip()) return + const store = makeStore('edgeempty') + await store.addDocuments([new Document({ pageContent: '', metadata: { empty: true } })]) + const results = await store.similaritySearchVectorWithScore(await new FakeEmbeddings().embedQuery(''), 1) + expect(results.length).toBe(1) + expect(results[0][0].pageContent).toBe('') + expect(results[0][0].metadata.empty).toBe(true) + }) + + it('should handle documents with very long content', async () => { + if (skip()) return + const store = makeStore('edgelong') + const longContent = 'x'.repeat(10000) + await store.addDocuments([new Document({ pageContent: longContent, metadata: {} })]) + const results = await store.similaritySearchVectorWithScore(await new FakeEmbeddings().embedQuery(longContent), 1) + expect(results.length).toBe(1) + expect(results[0][0].pageContent.length).toBe(10000) + }) + + it('should handle metadata with special characters', async () => { + if (skip()) return + const store = makeStore('edgespecial') + const meta = { key: 'value with "quotes" and {braces}', unicode: '日本語' } + await store.addDocuments([new Document({ pageContent: 'special chars', metadata: meta })]) + const results = await store.similaritySearchVectorWithScore(await new FakeEmbeddings().embedQuery('special chars'), 1) + expect(results[0][0].metadata).toEqual(meta) + }) + }) +}) diff --git a/packages/components/nodes/vectorstores/Valkey/Valkey.test.ts b/packages/components/nodes/vectorstores/Valkey/Valkey.test.ts new file mode 100644 index 00000000000..0d7f7d724f1 --- /dev/null +++ b/packages/components/nodes/vectorstores/Valkey/Valkey.test.ts @@ -0,0 +1,798 @@ +import { Document } from '@langchain/core/documents' + +// Mock @valkey/valkey-glide +const mockExec = jest.fn().mockResolvedValue([]) +const mockHset = jest.fn().mockReturnThis() +const mockExpire = jest.fn().mockReturnThis() +const mockDel = jest.fn().mockResolvedValue(1) +const mockScan = jest.fn().mockResolvedValue(['0', []]) +const mockClose = jest.fn() +const mockCustomCommand = jest.fn().mockResolvedValue([]) + +const mockGlideClient = { + exec: mockExec, + del: mockDel, + scan: mockScan, + close: mockClose, + customCommand: mockCustomCommand +} + +jest.mock('@valkey/valkey-glide', () => ({ + GlideClient: { + createClient: jest.fn().mockResolvedValue(mockGlideClient) + }, + GlideClusterClient: class GlideClusterClient {}, + ClusterScanCursor: class ClusterScanCursor { + _finished = true + constructor(finished = true) { + this._finished = finished + } + isFinished() { + return this._finished + } + }, + GlideFt: { + search: jest.fn().mockResolvedValue([0, []]), + info: jest.fn().mockResolvedValue({ numDocs: '0' }), + create: jest.fn().mockResolvedValue('OK'), + dropindex: jest.fn().mockResolvedValue('OK') + }, + Batch: jest.fn().mockImplementation(() => ({ + hset: mockHset, + expire: mockExpire + })), + ClusterBatch: jest.fn().mockImplementation(() => ({ + hset: mockHset, + expire: mockExpire + })), + Field: {} +})) + +const mockGetCredentialData = jest.fn().mockResolvedValue({}) +const mockGetCredentialParam = jest.fn() + +jest.mock('../../../src/utils', () => ({ + getBaseClasses: jest.fn(() => ['VectorStore']), + getCredentialData: (...args: any[]) => mockGetCredentialData(...args), + getCredentialParam: (...args: any[]) => mockGetCredentialParam(...args) +})) + +jest.mock('../../../src/indexing', () => ({ + index: jest.fn().mockResolvedValue({ numAdded: 1, addedDocs: [] }) +})) + +const { ValkeyVectorStore, parseConnectionUrl, nodeClass: Valkey_VectorStores, getConnectionConfig } = require('./Valkey') + +class FakeEmbeddings { + async embedDocuments(texts: string[]): Promise { + return texts.map(() => new Array(128).fill(0.1)) + } + async embedQuery(): Promise { + return new Array(128).fill(0.1) + } +} + +describe('parseConnectionUrl', () => { + it('should parse basic valkey URL', () => { + const config = parseConnectionUrl('valkey://localhost:6379') + expect(config).toEqual({ + host: 'localhost', + port: 6379, + username: undefined, + password: undefined, + useTLS: false + }) + }) + + it('should parse URL with credentials', () => { + const config = parseConnectionUrl('valkey://user:pass@myhost:6380') + expect(config.host).toBe('myhost') + expect(config.port).toBe(6380) + expect(config.username).toBe('user') + expect(config.password).toBe('pass') + }) + + it('should decode URL-encoded credentials', () => { + const config = parseConnectionUrl('valkey://user%40name:p%40ss%23word@host:6379') + expect(config.username).toBe('user@name') + expect(config.password).toBe('p@ss#word') + }) + + it('should detect TLS from valkeys:// protocol', () => { + const config = parseConnectionUrl('valkeys://host:6380') + expect(config.useTLS).toBe(true) + }) + + it('should detect TLS from rediss:// protocol', () => { + const config = parseConnectionUrl('rediss://host:6380') + expect(config.useTLS).toBe(true) + }) + + it('should not enable TLS for redis:// protocol', () => { + const config = parseConnectionUrl('redis://host:6379') + expect(config.useTLS).toBe(false) + }) + + it('should default port to 6379 when not specified', () => { + const config = parseConnectionUrl('valkey://host') + expect(config.port).toBe(6379) + }) + + it('should parse password-only auth (no username)', () => { + const config = parseConnectionUrl('valkey://:secret@localhost:6379') + expect(config.host).toBe('localhost') + expect(config.password).toBe('secret') + expect(config.username).toBeUndefined() + }) + + it('should handle password with percent-encoded special chars', () => { + const config = parseConnectionUrl('valkey://:%25percent%26amp@host:6379') + expect(config.password).toBe('%percent&') + }) + + it('should throw on invalid URL', () => { + expect(() => parseConnectionUrl('')).toThrow() + expect(() => parseConnectionUrl('not-a-url')).toThrow() + }) + + it('should default host to 127.0.0.1 when hostname is empty', () => { + // new URL('valkey://') parses with empty hostname + const config = parseConnectionUrl('valkey://') + expect(config.host).toBe('127.0.0.1') + }) +}) + +describe('ValkeyVectorStore', () => { + let store: any + + beforeEach(() => { + jest.clearAllMocks() + store = new ValkeyVectorStore(new FakeEmbeddings(), { + valkeyClient: mockGlideClient, + indexName: 'test-idx', + keyPrefix: 'doc:test:' + }) + }) + + describe('buildQuery', () => { + it('should produce wildcard query without filter', () => { + const [query] = store.buildQuery([0.1], 5, undefined) + expect(query).toBe('*=>[KNN 5 @content_vector $vector AS vector_score]') + }) + + it('should produce filter expression from string filter', () => { + const [query] = store.buildQuery([0.1], 3, 'mytag') + expect(query).toBe('@metadata_tags:{mytag}=>[KNN 3 @content_vector $vector AS vector_score]') + }) + + it('should produce OR expression from array filter', () => { + const [query] = store.buildQuery([0.1], 2, ['tag1', 'tag2']) + expect(query).toBe('@metadata_tags:{tag1 | tag2}=>[KNN 2 @content_vector $vector AS vector_score]') + }) + + it.each([ + ['tag:with,specials', 'tag\\:with\\,specials'], + ['tag{1}[2]', 'tag\\{1\\}\\[2\\]'], + ['hello world', 'hello\\ world'], + ["it's@here", "it\\'s\\@here"] + ])('should escape special characters in filter tag "%s"', (input, expected) => { + const [query] = store.buildQuery([0.1], 1, [input]) + expect(query).toContain(expected) + }) + + it('should fall back to wildcard query for empty filter array', () => { + const [query] = store.buildQuery([0.1], 5, []) + expect(query).toBe('*=>[KNN 5 @content_vector $vector AS vector_score]') + }) + + it('should return correct options structure', () => { + const [, options] = store.buildQuery([0.1, 0.2], 4, undefined) + expect(options.RETURN).toEqual(['metadata', 'content', 'vector_score']) + expect(options.SORTBY).toBe('vector_score') + expect(options.DIALECT).toBe(2) + expect(options.LIMIT).toEqual({ from: 0, size: 4 }) + expect(options.PARAMS.vector).toBeInstanceOf(Buffer) + }) + + it('should encode vector as Float32 buffer', () => { + const [, options] = store.buildQuery([1.0, 2.0], 1, undefined) + const buf = options.PARAMS.vector + const floats = new Float32Array(buf.buffer, buf.byteOffset, 2) + expect(floats[0]).toBeCloseTo(1.0) + expect(floats[1]).toBeCloseTo(2.0) + }) + }) + + describe('parseSearchResults', () => { + it('should return empty array for empty results', () => { + expect(store.parseSearchResults([0, []])).toEqual([]) + }) + + it('should return empty array for non-array input', () => { + expect(store.parseSearchResults(null)).toEqual([]) + expect(store.parseSearchResults(undefined)).toEqual([]) + expect(store.parseSearchResults('string')).toEqual([]) + }) + + it('should parse valid search results', () => { + const raw = [ + 1, + [ + { + value: [ + { key: 'content', value: 'hello world' }, + { key: 'metadata', value: '{"source":"test"}' }, + { key: 'vector_score', value: '0.5' } + ] + } + ] + ] + const results = store.parseSearchResults(raw) + expect(results.length).toBe(1) + expect(results[0][0]).toBeInstanceOf(Document) + expect(results[0][0].pageContent).toBe('hello world') + expect(results[0][0].metadata).toEqual({ source: 'test' }) + expect(results[0][1]).toBe(0.5) + }) + + it('should parse multiple documents correctly', () => { + const raw = [ + 3, + [ + { + value: [ + { key: 'content', value: 'first' }, + { key: 'metadata', value: '{"idx":1}' }, + { key: 'vector_score', value: '0.1' } + ] + }, + { + value: [ + { key: 'content', value: 'second' }, + { key: 'metadata', value: '{"idx":2}' }, + { key: 'vector_score', value: '0.3' } + ] + }, + { + value: [ + { key: 'content', value: 'third' }, + { key: 'metadata', value: '{"idx":3}' }, + { key: 'vector_score', value: '0.7' } + ] + } + ] + ] + const results = store.parseSearchResults(raw) + expect(results.length).toBe(3) + expect(results[0][0].pageContent).toBe('first') + expect(results[1][0].pageContent).toBe('second') + expect(results[2][0].pageContent).toBe('third') + expect(results[0][1]).toBe(0.1) + expect(results[2][1]).toBe(0.7) + }) + + it('should default to empty string when content field is missing', () => { + const raw = [ + 1, + [ + { + value: [ + { key: 'metadata', value: '{"a":1}' }, + { key: 'vector_score', value: '0.2' } + ] + } + ] + ] + const results = store.parseSearchResults(raw) + expect(results.length).toBe(1) + expect(results[0][0].pageContent).toBe('') + expect(results[0][0].metadata).toEqual({ a: 1 }) + }) + + it('should handle malformed metadata JSON gracefully', () => { + const raw = [ + 1, + [ + { + value: [ + { key: 'content', value: 'doc' }, + { key: 'metadata', value: 'not-json{' }, + { key: 'vector_score', value: '0.1' } + ] + } + ] + ] + const results = store.parseSearchResults(raw) + expect(results.length).toBe(1) + expect(results[0][0].metadata).toEqual({}) + }) + + it('should skip entries without vector_score', () => { + const raw = [ + 1, + [ + { + value: [ + { key: 'content', value: 'doc' }, + { key: 'metadata', value: '{}' } + ] + } + ] + ] + const results = store.parseSearchResults(raw) + expect(results.length).toBe(0) + }) + }) + + describe('delete', () => { + it('should no-op for empty ids array', async () => { + await store.delete({ ids: [] }) + expect(mockDel).not.toHaveBeenCalled() + }) + + it('should delete keys with prefix prepended when missing', async () => { + await store.delete({ ids: ['abc', 'def'] }) + expect(mockDel).toHaveBeenCalledWith(['doc:test:abc', 'doc:test:def']) + }) + + it('should not double-prefix keys that already have prefix', async () => { + await store.delete({ ids: ['doc:test:abc'] }) + expect(mockDel).toHaveBeenCalledWith(['doc:test:abc']) + }) + + it('should throw for invalid params', async () => { + await expect(store.delete({ invalid: true })).rejects.toThrow('Invalid parameters') + }) + + it('should drop index and scan/delete keys on deleteAll', async () => { + const { GlideFt } = require('@valkey/valkey-glide') + mockScan.mockResolvedValueOnce(['0', ['doc:test:key1', 'doc:test:key2']]) + await store.delete({ deleteAll: true }) + expect(GlideFt.dropindex).toHaveBeenCalledWith(mockGlideClient, 'test-idx') + expect(mockScan).toHaveBeenCalled() + expect(mockDel).toHaveBeenCalledWith(['doc:test:key1', 'doc:test:key2']) + }) + + it('should handle scan returning no keys on deleteAll', async () => { + mockScan.mockResolvedValueOnce(['0', []]) + await store.delete({ deleteAll: true }) + expect(mockDel).not.toHaveBeenCalled() + }) + + it('should use ClusterScanCursor for GlideClusterClient', async () => { + const { GlideClusterClient, ClusterScanCursor } = require('@valkey/valkey-glide') + const mockClusterScan = jest.fn().mockResolvedValue([new ClusterScanCursor(), ['doc:cluster:k1']]) + const mockClusterDel = jest.fn().mockResolvedValue(1) + const clusterClient = Object.create(GlideClusterClient.prototype) + clusterClient.scan = mockClusterScan + clusterClient.del = mockClusterDel + clusterClient.exec = jest.fn() + + const clusterStore = new ValkeyVectorStore(new FakeEmbeddings(), { + valkeyClient: clusterClient, + indexName: 'cluster-idx', + keyPrefix: 'doc:cluster:' + }) + + await clusterStore.delete({ deleteAll: true }) + expect(mockClusterScan).toHaveBeenCalled() + expect(mockClusterDel).toHaveBeenCalledWith(['doc:cluster:k1']) + }) + }) + + describe('addVectors', () => { + it('should throw when no vectors provided', async () => { + await expect(store.addVectors([], [])).rejects.toThrow('No vectors provided') + }) + + it('should throw when vectors are empty arrays', async () => { + await expect(store.addVectors([[]], [new Document({ pageContent: '', metadata: {} })])).rejects.toThrow('No vectors provided') + }) + + it('should throw when vectors.length !== documents.length', async () => { + await expect( + store.addVectors( + [ + [0.1, 0.2], + [0.3, 0.4] + ], + [new Document({ pageContent: 'only one', metadata: {} })] + ) + ).rejects.toThrow('Vectors length (2) must match documents length (1)') + }) + + it('should generate UUID-based keys', async () => { + const { GlideFt } = require('@valkey/valkey-glide') + GlideFt.info.mockResolvedValue({ numDocs: '0' }) + + await store.addVectors([[0.1, 0.2]], [new Document({ pageContent: 'test', metadata: {} })]) + + const { Batch } = require('@valkey/valkey-glide') + const batchInstance = Batch.mock.results[0].value + const hsetCall = batchInstance.hset.mock.calls[0] + const key = hsetCall[0] + expect(key).toMatch(/^doc:test:[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/) + }) + + it('should use user-provided keys when given', async () => { + await store.addVectors([[0.1, 0.2]], [new Document({ pageContent: 'test', metadata: {} })], { + keys: ['my-custom-key'] + }) + + const { Batch } = require('@valkey/valkey-glide') + const batchInstance = Batch.mock.results[0].value + expect(batchInstance.hset).toHaveBeenCalledWith('my-custom-key', expect.any(Object)) + }) + + it('should set TTL when configured', async () => { + const storeWithTTL = new ValkeyVectorStore(new FakeEmbeddings(), { + valkeyClient: mockGlideClient, + indexName: 'test-idx', + keyPrefix: 'doc:test:', + ttl: 300 + }) + + await storeWithTTL.addVectors([[0.1, 0.2]], [new Document({ pageContent: 'test', metadata: {} })]) + + const { Batch } = require('@valkey/valkey-glide') + const batchInstance = Batch.mock.results[0].value + expect(batchInstance.expire).toHaveBeenCalled() + expect(batchInstance.expire.mock.calls[0][1]).toBe(300) + }) + + it('should not set TTL when not configured', async () => { + mockExpire.mockClear() + await store.addVectors([[0.1, 0.2]], [new Document({ pageContent: 'test', metadata: {} })]) + + const { Batch } = require('@valkey/valkey-glide') + const batchInstance = Batch.mock.results[0].value + expect(batchInstance.expire).not.toHaveBeenCalled() + }) + + it('should flush batch at exactly batchSize boundary', async () => { + const docs = Array.from({ length: 3 }, (_, i) => new Document({ pageContent: `doc${i}`, metadata: {} })) + const vecs = Array.from({ length: 3 }, () => [0.1, 0.2]) + + await store.addVectors(vecs, docs, { batchSize: 3 }) + + // Exactly one batch exec call (3 docs, batchSize=3) + expect(mockExec).toHaveBeenCalledTimes(1) + }) + + it('should split into multiple batches when exceeding batchSize', async () => { + const docs = Array.from({ length: 5 }, (_, i) => new Document({ pageContent: `doc${i}`, metadata: {} })) + const vecs = Array.from({ length: 5 }, () => [0.1, 0.2]) + + await store.addVectors(vecs, docs, { batchSize: 2 }) + + // 5 docs with batchSize=2 → 3 exec calls (2, 2, 1) + expect(mockExec).toHaveBeenCalledTimes(3) + }) + + it('should propagate error when exec fails mid-batch', async () => { + mockExec.mockRejectedValueOnce(new Error('Connection lost')) + + const docs = Array.from({ length: 3 }, (_, i) => new Document({ pageContent: `doc${i}`, metadata: {} })) + const vecs = Array.from({ length: 3 }, () => [0.1, 0.2]) + + await expect(store.addVectors(vecs, docs, { batchSize: 2 })).rejects.toThrow('Connection lost') + }) + }) + + describe('TTL validation', () => { + it('should throw for ttl: 0', () => { + expect( + () => + new ValkeyVectorStore(new FakeEmbeddings(), { + valkeyClient: mockGlideClient, + indexName: 'test-idx', + ttl: 0 + }) + ).toThrow('TTL must be a positive integer') + }) + + it('should throw for negative TTL', () => { + expect( + () => + new ValkeyVectorStore(new FakeEmbeddings(), { + valkeyClient: mockGlideClient, + indexName: 'test-idx', + ttl: -10 + }) + ).toThrow('TTL must be a positive integer') + }) + + it('should accept undefined TTL (no expiry)', () => { + expect( + () => + new ValkeyVectorStore(new FakeEmbeddings(), { + valkeyClient: mockGlideClient, + indexName: 'test-idx' + }) + ).not.toThrow() + }) + }) + + describe('createIndex', () => { + it('should not recreate existing index', async () => { + const { GlideFt } = require('@valkey/valkey-glide') + GlideFt.info.mockResolvedValue({ numDocs: '0' }) + await store.createIndex(128) + expect(GlideFt.create).not.toHaveBeenCalled() + }) + + it('should create index with TAG and VECTOR fields', async () => { + const { GlideFt } = require('@valkey/valkey-glide') + GlideFt.info.mockRejectedValueOnce(new Error('Unknown index')) + await store.createIndex(128) + expect(GlideFt.create).toHaveBeenCalledWith( + mockGlideClient, + 'test-idx', + expect.arrayContaining([ + expect.objectContaining({ type: 'TAG', name: 'metadata_tags' }), + expect.objectContaining({ type: 'VECTOR', name: 'content_vector' }) + ]), + expect.objectContaining({ dataType: 'HASH', prefixes: ['doc:test:'] }) + ) + }) + + it('should throw helpful error when TEXT field unsupported', async () => { + const { GlideFt } = require('@valkey/valkey-glide') + GlideFt.info.mockRejectedValueOnce(new Error('Unknown index')) + GlideFt.create.mockRejectedValueOnce(new Error('unknown field type TEXT')) + await expect(store.createIndex(128)).rejects.toThrow('Valkey Search >= 1.2') + }) + + it('should handle concurrent index creation (already exists)', async () => { + const { GlideFt } = require('@valkey/valkey-glide') + GlideFt.info.mockRejectedValueOnce(new Error('Unknown index')) + GlideFt.create.mockRejectedValueOnce(new Error('Index already exists')) + // Should not throw — handled gracefully + await expect(store.createIndex(128)).resolves.toBeUndefined() + }) + }) + + describe('checkIndexExists', () => { + it('should return true when index exists', async () => { + const { GlideFt } = require('@valkey/valkey-glide') + GlideFt.info.mockResolvedValue({ numDocs: '0' }) + expect(await store.checkIndexExists()).toBe(true) + }) + + it('should return false when index does not exist', async () => { + const { GlideFt } = require('@valkey/valkey-glide') + GlideFt.info.mockRejectedValueOnce(new Error('Unknown index name')) + expect(await store.checkIndexExists()).toBe(false) + }) + + it('should throw when search module not loaded', async () => { + const { GlideFt } = require('@valkey/valkey-glide') + GlideFt.info.mockRejectedValueOnce(new Error('unknown command FT.INFO')) + await expect(store.checkIndexExists()).rejects.toThrow('valkey-search module') + }) + }) +}) + +describe('Valkey_VectorStores node wrapper', () => { + let node: any + + beforeEach(() => { + jest.clearAllMocks() + node = new Valkey_VectorStores() + mockGetCredentialParam.mockImplementation((key: string) => { + const map: Record = { + valkeyUrl: 'valkey://localhost:6379', + valkeyHost: '', + valkeyPort: '', + valkeyUser: '', + valkeyPassword: '', + valkeyTls: '' + } + return map[key] || '' + }) + }) + + describe('init', () => { + it('should create a vector store and return retriever when output is retriever', async () => { + const nodeData = { + credential: 'cred-id', + inputs: { + indexName: 'my-index', + embeddings: new FakeEmbeddings(), + topK: '5', + contentKey: 'content', + metadataKey: 'metadata', + vectorKey: 'content_vector', + valkeyMetadataFilter: undefined + }, + outputs: { output: 'retriever' } + } + + const result = await node.init(nodeData, '', {}) + // Retriever has invoke method + expect(result).toBeDefined() + expect(result.lc_namespace).toBeDefined() + }) + + it('should return vectorStore when output is vectorStore', async () => { + const nodeData = { + credential: 'cred-id', + inputs: { + indexName: 'my-index', + embeddings: new FakeEmbeddings(), + topK: '3', + contentKey: '', + metadataKey: '', + vectorKey: '', + valkeyMetadataFilter: undefined + }, + outputs: { output: 'vectorStore' } + } + + const result = await node.init(nodeData, '', {}) + expect(result._vectorstoreType()).toBe('valkey') + expect((result as any).k).toBe(3) + }) + + it('should parse JSON metadata filter', async () => { + const nodeData = { + credential: 'cred-id', + inputs: { + indexName: 'my-index', + embeddings: new FakeEmbeddings(), + topK: '4', + contentKey: '', + metadataKey: '', + vectorKey: '', + valkeyMetadataFilter: '["tag1","tag2"]' + }, + outputs: { output: 'vectorStore' } + } + + const result = await node.init(nodeData, '', {}) + expect(result.filter).toEqual(['tag1', 'tag2']) + }) + }) + + describe('vectorStoreMethods.upsert', () => { + it('should upsert documents and close client', async () => { + const nodeData = { + credential: 'cred-id', + inputs: { + indexName: 'upsert-idx', + embeddings: new FakeEmbeddings(), + replaceIndex: false, + contentKey: 'content', + metadataKey: 'metadata', + vectorKey: 'content_vector', + recordManager: undefined, + document: [[new Document({ pageContent: 'hello', metadata: {} })]] + } + } + + const result = await node.vectorStoreMethods.upsert(nodeData, {}) + expect(result.numAdded).toBe(1) + expect(mockClose).toHaveBeenCalled() + }) + + it('should drop index when replaceIndex is true', async () => { + const { GlideFt } = require('@valkey/valkey-glide') + const nodeData = { + credential: 'cred-id', + inputs: { + indexName: 'replace-idx', + embeddings: new FakeEmbeddings(), + replaceIndex: true, + contentKey: '', + metadataKey: '', + vectorKey: '', + recordManager: undefined, + document: [[new Document({ pageContent: 'test', metadata: {} })]] + } + } + + await node.vectorStoreMethods.upsert(nodeData, {}) + expect(GlideFt.dropindex).toHaveBeenCalled() + }) + + it('should filter out documents with empty pageContent', async () => { + const nodeData = { + credential: 'cred-id', + inputs: { + indexName: 'filter-idx', + embeddings: new FakeEmbeddings(), + replaceIndex: false, + contentKey: '', + metadataKey: '', + vectorKey: '', + recordManager: undefined, + document: [[new Document({ pageContent: 'keep', metadata: {} }), { pageContent: '', metadata: {} }, null]] + } + } + + const result = await node.vectorStoreMethods.upsert(nodeData, {}) + expect(result.numAdded).toBe(1) + }) + + it('should close client even on error', async () => { + mockExec.mockRejectedValueOnce(new Error('exec failed')) + const { GlideFt } = require('@valkey/valkey-glide') + GlideFt.info.mockRejectedValueOnce(new Error('Unknown index')) + + const nodeData = { + credential: 'cred-id', + inputs: { + indexName: 'error-idx', + embeddings: new FakeEmbeddings(), + replaceIndex: false, + contentKey: '', + metadataKey: '', + vectorKey: '', + recordManager: undefined, + document: [[new Document({ pageContent: 'test', metadata: {} })]] + } + } + + await expect(node.vectorStoreMethods.upsert(nodeData, {})).rejects.toThrow() + expect(mockClose).toHaveBeenCalled() + }) + }) + + describe('vectorStoreMethods.delete', () => { + it('should delete by ids and close client', async () => { + const nodeData = { + credential: 'cred-id', + inputs: { + indexName: 'del-idx', + embeddings: new FakeEmbeddings(), + contentKey: '', + metadataKey: '', + vectorKey: '', + recordManager: undefined + } + } + + await node.vectorStoreMethods.delete(nodeData, ['key1', 'key2'], {}) + expect(mockDel).toHaveBeenCalled() + expect(mockClose).toHaveBeenCalled() + }) + }) +}) + +describe('getConnectionConfig', () => { + beforeEach(() => { + jest.clearAllMocks() + }) + + it('should use URL when valkeyUrl is provided', () => { + mockGetCredentialParam.mockImplementation((key: string) => { + if (key === 'valkeyUrl') return 'valkey://myhost:6380' + return '' + }) + + const config = getConnectionConfig({}, { inputs: {} }) + expect(config.host).toBe('myhost') + expect(config.port).toBe(6380) + }) + + it('should fall back to individual fields when URL is empty', () => { + mockGetCredentialParam.mockImplementation((key: string) => { + const map: Record = { + valkeyUrl: '', + valkeyHost: '10.0.0.1', + valkeyPort: '6381', + valkeyUser: 'admin', + valkeyPassword: 'secret', + valkeyTls: 'true' + } + return map[key] || '' + }) + + const config = getConnectionConfig({}, { inputs: {} }) + expect(config.host).toBe('10.0.0.1') + expect(config.port).toBe(6381) + expect(config.username).toBe('admin') + expect(config.password).toBe('secret') + expect(config.useTLS).toBe(true) + }) +}) diff --git a/packages/components/nodes/vectorstores/Valkey/Valkey.ts b/packages/components/nodes/vectorstores/Valkey/Valkey.ts new file mode 100644 index 00000000000..b1cd534f1aa --- /dev/null +++ b/packages/components/nodes/vectorstores/Valkey/Valkey.ts @@ -0,0 +1,698 @@ +import { flatten } from 'lodash' +import { randomUUID } from 'crypto' +import { Batch, ClusterBatch, ClusterScanCursor, GlideClient, GlideClusterClient, GlideFt, Field } from '@valkey/valkey-glide' +import { VectorStore } from '@langchain/core/vectorstores' +import type { EmbeddingsInterface } from '@langchain/core/embeddings' +import { Embeddings } from '@langchain/core/embeddings' +import { Document } from '@langchain/core/documents' +import { ICommonObject, INode, INodeData, INodeOutputsValue, INodeParams, IndexingResult } from '../../../src/Interface' +import { getBaseClasses, getCredentialData, getCredentialParam } from '../../../src/utils' +import { index } from '../../../src/indexing' + +// ValkeyVectorStore — adapted from @langchain/valkey (langchainjs) +// https://github.com/daric93/langchainjs/tree/feature/valkey-vector-store + +export enum VectorAlgorithms { + FLAT = 'FLAT', + HNSW = 'HNSW' +} + +export type CreateSchemaVectorField> = { + ALGORITHM: T + DISTANCE_METRIC: 'L2' | 'IP' | 'COSINE' + INITIAL_CAP?: number +} & A + +export type CreateSchemaFlatVectorField = CreateSchemaVectorField + +export type CreateSchemaHNSWVectorField = CreateSchemaVectorField< + VectorAlgorithms.HNSW, + { M?: number; EF_CONSTRUCTION?: number; EF_RUNTIME?: number } +> + +export interface ValkeyVectorStoreConfig { + valkeyClient: GlideClient | GlideClusterClient + indexName: string + indexOptions?: CreateSchemaFlatVectorField | CreateSchemaHNSWVectorField + createIndexOptions?: Record + keyPrefix?: string + contentKey?: string + metadataKey?: string + vectorKey?: string + filter?: ValkeyVectorStoreFilterType + ttl?: number +} + +export interface ValkeyAddOptions { + keys?: string[] + batchSize?: number +} + +export type ValkeyVectorStoreFilterType = string[] | string + +export class ValkeyVectorStore extends VectorStore { + declare FilterType: ValkeyVectorStoreFilterType + + valkeyClient: GlideClient | GlideClusterClient + + indexName: string + + indexOptions: CreateSchemaFlatVectorField | CreateSchemaHNSWVectorField + + createIndexOptions: Record + + keyPrefix: string + + contentKey: string + + metadataKey: string + + vectorKey: string + + filter?: ValkeyVectorStoreFilterType + + ttl?: number + + _vectorstoreType(): string { + return 'valkey' + } + + constructor(embeddings: EmbeddingsInterface, _dbConfig: ValkeyVectorStoreConfig) { + super(embeddings, _dbConfig) + this.valkeyClient = _dbConfig.valkeyClient + this.indexName = _dbConfig.indexName + this.indexOptions = _dbConfig.indexOptions ?? { + ALGORITHM: VectorAlgorithms.HNSW, + DISTANCE_METRIC: 'COSINE' + } + this.keyPrefix = _dbConfig.keyPrefix ?? `doc:${this.indexName}:` + this.contentKey = _dbConfig.contentKey ?? 'content' + this.metadataKey = _dbConfig.metadataKey ?? 'metadata' + this.vectorKey = _dbConfig.vectorKey ?? 'content_vector' + this.filter = _dbConfig.filter + if (_dbConfig.ttl !== undefined && _dbConfig.ttl <= 0) { + throw new Error(`TTL must be a positive integer, got ${_dbConfig.ttl}`) + } + this.ttl = _dbConfig.ttl + this.createIndexOptions = { + ON: 'HASH', + PREFIX: this.keyPrefix, + ...(_dbConfig.createIndexOptions || {}) + } + } + + async checkIndexExists(): Promise { + try { + await GlideFt.info(this.valkeyClient, this.indexName) + return true + } catch (err) { + if ((err as Error)?.message.includes('unknown command')) { + throw new Error( + 'Failed to run FT.INFO command. Please ensure that your Valkey instance has the valkey-search module enabled.' + ) + } + return false + } + } + + async createIndex(dimensions = 1536): Promise { + if (await this.checkIndexExists()) { + return + } + + // TAG field requires Valkey Search >= 1.2 (Valkey >= 9.1) + const schema: Field[] = [ + { + type: 'TAG', + name: `${this.metadataKey}_tags`, + separator: ',' + }, + { + type: 'VECTOR', + name: this.vectorKey, + attributes: { + algorithm: this.indexOptions.ALGORITHM, + type: 'FLOAT32', + dimensions: dimensions, + distanceMetric: this.indexOptions.DISTANCE_METRIC + } + } + ] + + await GlideFt.create(this.valkeyClient, this.indexName, schema, { + dataType: 'HASH', + prefixes: [this.keyPrefix] + }).catch((err) => { + const msg = (err as Error)?.message || '' + if (msg.includes('already exists')) { + return // Index was created concurrently — safe to continue + } + if (msg.includes('TAG') || msg.includes('unsupported') || msg.includes('unknown field type')) { + throw new Error( + 'Failed to create index with TAG field. Metadata filtering requires Valkey Search >= 1.2 (Valkey >= 9.1). ' + + `Original error: ${msg}` + ) + } + throw err + }) + } + + async dropIndex(): Promise { + try { + await GlideFt.dropindex(this.valkeyClient, this.indexName) + return true + } catch { + return false + } + } + + async delete(params: { deleteAll: boolean } | { ids: string[] }): Promise { + if ('ids' in params && params.ids?.length === 0) return + if ('deleteAll' in params && params.deleteAll) { + await this.dropIndex() + // Scan and delete orphaned HASH keys + if (this.valkeyClient instanceof GlideClusterClient) { + let cursor = new ClusterScanCursor() + do { + const [next, keys] = await this.valkeyClient.scan(cursor, { match: `${this.keyPrefix}*`, count: 100 }) + cursor = next + if (keys.length) await this.valkeyClient.del(keys as string[]) + } while (!cursor.isFinished()) + } else { + let cursor = '0' + do { + const result = await (this.valkeyClient as GlideClient).scan(cursor, { + match: `${this.keyPrefix}*`, + count: 100 + }) + cursor = String(result[0]) + const keys = result[1] as string[] + if (keys.length) await this.valkeyClient.del(keys) + } while (cursor !== '0') + } + } else if ('ids' in params && params.ids && params.ids.length > 0) { + const keys = params.ids.map((id) => (id.startsWith(this.keyPrefix) ? id : `${this.keyPrefix}${id}`)) + await this.valkeyClient.del(keys) + } else { + throw new Error(`Invalid parameters passed to "delete".`) + } + } + + async addDocuments(documents: Document[], options?: ValkeyAddOptions) { + const texts = documents.map(({ pageContent }) => pageContent) + return this.addVectors(await this.embeddings.embedDocuments(texts), documents, options) + } + + async addVectors(vectors: number[][], documents: Document[], { keys, batchSize = 100 }: ValkeyAddOptions = {}) { + if (!vectors.length || !vectors[0].length) { + throw new Error('No vectors provided') + } + if (vectors.length !== documents.length) { + throw new Error(`Vectors length (${vectors.length}) must match documents length (${documents.length})`) + } + await this.createIndex(vectors[0].length) + + const commands: Array<{ key: string; fields: Record }> = [] + + for (let idx = 0; idx < vectors.length; idx += 1) { + const vector = vectors[idx] + const key = keys && keys.length ? keys[idx] : `${this.keyPrefix}${randomUUID()}` + const metadata = documents[idx] && documents[idx].metadata ? documents[idx].metadata : {} + + const hashFields: Record = { + [this.vectorKey]: this.getFloat32Buffer(vector), + [this.contentKey]: documents[idx]?.pageContent || '', + [this.metadataKey]: JSON.stringify(metadata), + [`${this.metadataKey}_tags`]: Object.values(metadata) + .filter((v) => v != null && typeof v !== 'object') + .map((v) => String(v)) + .join(',') + } + + commands.push({ key, fields: hashFields }) + + if (commands.length >= batchSize || idx === vectors.length - 1) { + if (this.valkeyClient instanceof GlideClusterClient) { + const batch = new ClusterBatch(false) // non-atomic: allows multi-slot routing + for (const { key, fields } of commands) { + batch.hset(key, fields) + if (this.ttl) { + batch.expire(key, this.ttl) + } + } + await this.valkeyClient.exec(batch, true, { retryStrategy: { retryServerError: true, retryConnectionError: true } }) + } else { + const batch = new Batch(false) + for (const { key, fields } of commands) { + batch.hset(key, fields) + if (this.ttl) { + batch.expire(key, this.ttl) + } + } + await (this.valkeyClient as GlideClient).exec(batch, true) + } + commands.length = 0 + } + } + } + + async similaritySearchVectorWithScore(query: number[], k: number, filter?: ValkeyVectorStoreFilterType): Promise<[Document, number][]> { + if (filter && this.filter) { + throw new Error('cannot provide both `filter` and `this.filter`') + } + + if (!(await this.checkIndexExists())) { + return [] + } + + const _filter = filter ?? this.filter + const [queryStr, options] = this.buildQuery(query, k, _filter) + + const searchOptions = { + params: [{ key: 'vector', value: options.PARAMS.vector }], + returnFields: options.RETURN.map((field) => ({ fieldIdentifier: field })) + } + + const results = await GlideFt.search(this.valkeyClient, this.indexName, queryStr, searchOptions) + return this.parseSearchResults(results) + } + + private parseSearchResults(results: unknown): [Document, number][] { + const result: [Document, number][] = [] + + if (Array.isArray(results) && results.length > 1) { + const documents = results[1] + if (Array.isArray(documents)) { + for (const doc of documents) { + if (Array.isArray(doc?.value)) { + const fieldsObj: Record = {} + for (const field of doc.value) { + if (field?.key !== undefined && field?.value !== undefined) { + fieldsObj[String(field.key)] = field.value + } + } + + if (fieldsObj.vector_score !== undefined) { + let metadata: Record = {} + try { + metadata = JSON.parse((fieldsObj[this.metadataKey] ?? '{}') as string) + } catch { + metadata = {} + } + + result.push([ + new Document({ + pageContent: (fieldsObj[this.contentKey] ?? '') as string, + metadata + }), + Number(fieldsObj.vector_score) + ]) + } + } + } + } + } + + return result + } + + private buildQuery( + query: number[], + k: number, + _filter?: ValkeyVectorStoreFilterType + ): [string, { PARAMS: { vector: Buffer }; RETURN: string[]; SORTBY: string; DIALECT: number; LIMIT: { from: number; size: number } }] { + const vectorScoreField = 'vector_score' + let hybridFields = '*' + + if (_filter) { + const tags = Array.isArray(_filter) ? _filter : [_filter] + if (tags.length > 0) { + const escaped = tags.map((t) => t.replace(/\\/g, '\\\\').replace(/[,.<>{}[\]"':;!@#$%^&*()\-+=~|/ ]/g, '\\$&')) + hybridFields = `@${this.metadataKey}_tags:{${escaped.join(' | ')}}` + } + } + + const baseQuery = `${hybridFields}=>[KNN ${k} @${this.vectorKey} $vector AS ${vectorScoreField}]` + const returnFields = [this.metadataKey, this.contentKey, vectorScoreField] + + const options = { + PARAMS: { vector: this.getFloat32Buffer(query) }, + RETURN: returnFields, + SORTBY: vectorScoreField, + DIALECT: 2, + LIMIT: { from: 0, size: k } + } + + return [baseQuery, options] + } + + private getFloat32Buffer(vector: number[]): Buffer { + return Buffer.from(new Float32Array(vector).buffer) + } + + static async fromDocuments( + docs: Document[], + embeddings: EmbeddingsInterface, + dbConfig: ValkeyVectorStoreConfig, + docsOptions?: ValkeyAddOptions + ): Promise { + const instance = new this(embeddings, dbConfig) + await instance.addDocuments(docs, docsOptions) + return instance + } + + static fromTexts( + texts: string[], + metadatas: object[] | object, + embeddings: EmbeddingsInterface, + dbConfig: ValkeyVectorStoreConfig, + docsOptions?: ValkeyAddOptions + ): Promise { + const docs: Document[] = [] + for (let i = 0; i < texts.length; i += 1) { + const metadata = Array.isArray(metadatas) ? metadatas[i] : metadatas + docs.push(new Document({ pageContent: texts[i], metadata })) + } + return ValkeyVectorStore.fromDocuments(docs, embeddings, dbConfig, docsOptions) + } +} + +// Flowise Node Wrapper + +interface ValkeyConnectionConfig { + host: string + port: number + username?: string + password?: string + useTLS?: boolean +} + +function parseConnectionUrl(url: string): ValkeyConnectionConfig { + const parsed = new URL(url) + return { + host: parsed.hostname || '127.0.0.1', + port: parseInt(parsed.port, 10) || 6379, + username: parsed.username ? decodeURIComponent(parsed.username) : undefined, + password: parsed.password ? decodeURIComponent(parsed.password) : undefined, + useTLS: parsed.protocol === 'rediss:' || parsed.protocol === 'valkeys:' + } +} + +async function createGlideClient(config: ValkeyConnectionConfig): Promise { + const clientConfig: any = { + addresses: [{ host: config.host, port: config.port }], + requestTimeout: 5000, + connectionTimeout: 2000 + } + if (config.username || config.password) { + clientConfig.credentials = { + ...(config.username && { username: config.username }), + ...(config.password && { password: config.password }) + } + } + if (config.useTLS) { + clientConfig.useTLS = true + } + return await GlideClient.createClient(clientConfig) +} + +function getConnectionConfig(credentialData: ICommonObject, nodeData: INodeData): ValkeyConnectionConfig { + const url = getCredentialParam('valkeyUrl', credentialData, nodeData) + if (url && url !== '') { + return parseConnectionUrl(url) + } + const host = getCredentialParam('valkeyHost', credentialData, nodeData) || '127.0.0.1' + const port = parseInt(getCredentialParam('valkeyPort', credentialData, nodeData) || '6379', 10) + const username = getCredentialParam('valkeyUser', credentialData, nodeData) || undefined + const password = getCredentialParam('valkeyPassword', credentialData, nodeData) || undefined + const useTLS = getCredentialParam('valkeyTls', credentialData, nodeData) === 'true' + return { host, port, username, password, useTLS } +} + +class Valkey_VectorStores implements INode { + label: string + name: string + version: number + description: string + type: string + icon: string + category: string + badge: string + baseClasses: string[] + inputs: INodeParams[] + credential: INodeParams + outputs: INodeOutputsValue[] + + constructor() { + this.label = 'Valkey' + this.name = 'valkey' + this.version = 1.0 + this.description = + 'Upsert embedded data and perform similarity search upon query using Valkey, a high-performance open-source key-value store. Requires Valkey Search >= 1.2 (Valkey >= 9.1) for metadata filtering.' + this.type = 'Valkey' + this.icon = 'valkey.svg' + this.category = 'Vector Stores' + this.badge = 'NEW' + this.baseClasses = [this.type, 'VectorStoreRetriever', 'BaseRetriever'] + this.credential = { + label: 'Connect Credential', + name: 'credential', + type: 'credential', + credentialNames: ['valkeyUrlApi', 'valkeyApi'] + } + this.inputs = [ + { + label: 'Document', + name: 'document', + type: 'Document', + list: true, + optional: true + }, + { + label: 'Embeddings', + name: 'embeddings', + type: 'Embeddings' + }, + { + label: 'Record Manager', + name: 'recordManager', + type: 'RecordManager', + description: 'Keep track of the record to prevent duplication', + optional: true + }, + { + label: 'Index Name', + name: 'indexName', + description: + 'Name of the Valkey search index used for vector similarity queries. Will be created automatically on upsert if it does not exist.', + placeholder: '', + type: 'string' + }, + { + label: 'Replace Index on Upsert', + name: 'replaceIndex', + description: 'Selecting this option will delete the existing index and recreate a new one when upserting', + default: false, + type: 'boolean' + }, + { + label: 'Content Field', + name: 'contentKey', + description: 'Name of the field (column) that contains the actual content', + type: 'string', + default: 'content', + additionalParams: true, + optional: true + }, + { + label: 'Metadata Field', + name: 'metadataKey', + description: 'Name of the field (column) that contains the metadata of the document', + type: 'string', + default: 'metadata', + additionalParams: true, + optional: true + }, + { + label: 'Vector Field', + name: 'vectorKey', + description: 'Name of the field (column) that contains the vector', + type: 'string', + default: 'content_vector', + additionalParams: true, + optional: true + }, + { + label: 'Valkey Metadata Filter', + name: 'valkeyMetadataFilter', + type: 'json', + description: + 'Filter documents by metadata tags. Provide a JSON array of strings to match against the metadata field using OR logic, e.g. ["tag1", "tag2"]', + optional: true, + additionalParams: true, + acceptVariable: true + }, + { + label: 'Top K', + name: 'topK', + description: 'Number of top results to fetch. Default to 4', + placeholder: '4', + type: 'number', + additionalParams: true, + optional: true + } + ] + this.outputs = [ + { + label: 'Valkey Retriever', + name: 'retriever', + baseClasses: this.baseClasses + }, + { + label: 'Valkey Vector Store', + name: 'vectorStore', + baseClasses: [this.type, ...getBaseClasses(ValkeyVectorStore)] + } + ] + } + + //@ts-ignore + vectorStoreMethods = { + async upsert(nodeData: INodeData, options: ICommonObject): Promise> { + const credentialData = await getCredentialData(nodeData.credential ?? '', options) + const connectionConfig = getConnectionConfig(credentialData, nodeData) + const indexName = nodeData.inputs?.indexName as string + const embeddings = nodeData.inputs?.embeddings as Embeddings + const replaceIndex = nodeData.inputs?.replaceIndex as boolean + const contentKey = (nodeData.inputs?.contentKey as string) || 'content' + const metadataKey = (nodeData.inputs?.metadataKey as string) || 'metadata' + const vectorKey = (nodeData.inputs?.vectorKey as string) || 'content_vector' + const recordManager = nodeData.inputs?.recordManager + + const docs = nodeData.inputs?.document as Document[] + const flattenDocs = docs && docs.length ? flatten(docs) : [] + const finalDocs: Document[] = [] + for (let i = 0; i < flattenDocs.length; i += 1) { + if (flattenDocs[i] && flattenDocs[i].pageContent) { + finalDocs.push(new Document(flattenDocs[i])) + } + } + + const client = await createGlideClient(connectionConfig) + try { + const vectorStore = new ValkeyVectorStore(embeddings, { + valkeyClient: client, + indexName, + contentKey, + metadataKey, + vectorKey + }) + + if (replaceIndex) { + await vectorStore.dropIndex() + } + + if (recordManager) { + await recordManager.createSchema() + const res = await index({ + docsSource: finalDocs, + recordManager, + vectorStore, + options: { + cleanup: recordManager?.cleanup, + sourceIdKey: recordManager?.sourceIdKey ?? 'source', + vectorStoreName: indexName + } + }) + return res + } else { + await vectorStore.addDocuments(finalDocs) + return { numAdded: finalDocs.length, addedDocs: finalDocs } + } + } catch (e) { + throw e instanceof Error ? e : new Error(String(e)) + } finally { + client.close() + } + }, + async delete(nodeData: INodeData, ids: string[], options: ICommonObject): Promise { + const credentialData = await getCredentialData(nodeData.credential ?? '', options) + const connectionConfig = getConnectionConfig(credentialData, nodeData) + const indexName = nodeData.inputs?.indexName as string + const embeddings = nodeData.inputs?.embeddings as Embeddings + const contentKey = (nodeData.inputs?.contentKey as string) || 'content' + const metadataKey = (nodeData.inputs?.metadataKey as string) || 'metadata' + const vectorKey = (nodeData.inputs?.vectorKey as string) || 'content_vector' + const recordManager = nodeData.inputs?.recordManager + + const client = await createGlideClient(connectionConfig) + try { + const vectorStore = new ValkeyVectorStore(embeddings, { + valkeyClient: client, + indexName, + contentKey, + metadataKey, + vectorKey + }) + + if (recordManager) { + const vectorStoreName = indexName + await recordManager.createSchema() + ;(recordManager as any).namespace = (recordManager as any).namespace + '_' + vectorStoreName + const keys: string[] = await recordManager.listKeys({}) + await vectorStore.delete({ ids: keys }) + await recordManager.deleteKeys(keys) + } else { + await vectorStore.delete({ ids }) + } + } catch (e) { + throw e instanceof Error ? e : new Error(String(e)) + } finally { + client.close() + } + } + } + + async init(nodeData: INodeData, _: string, options: ICommonObject): Promise { + const credentialData = await getCredentialData(nodeData.credential ?? '', options) + const connectionConfig = getConnectionConfig(credentialData, nodeData) + const indexName = nodeData.inputs?.indexName as string + const embeddings = nodeData.inputs?.embeddings as Embeddings + const topK = nodeData.inputs?.topK as string + const k = topK ? Math.max(1, parseInt(topK, 10)) : 4 + const output = nodeData.outputs?.output as string + const contentKey = (nodeData.inputs?.contentKey as string) || 'content' + const metadataKey = (nodeData.inputs?.metadataKey as string) || 'metadata' + const vectorKey = (nodeData.inputs?.vectorKey as string) || 'content_vector' + const valkeyMetadataFilter = nodeData.inputs?.valkeyMetadataFilter + + let filter: ValkeyVectorStoreFilterType | undefined + if (valkeyMetadataFilter) { + filter = typeof valkeyMetadataFilter === 'object' ? valkeyMetadataFilter : JSON.parse(valkeyMetadataFilter) + } + + // Persistent client for the vector store's lifetime. Matches Postgres/TypeORM pattern in + // Flowise — no framework teardown hook exists. Recreating per query would forfeit GLIDE's + // multiplexed connection and automatic reconnection. One TCP connection per chatflow instance. + const client = await createGlideClient(connectionConfig) + + const vectorStore = new ValkeyVectorStore(embeddings, { + valkeyClient: client, + indexName, + contentKey, + metadataKey, + vectorKey, + filter + }) + + if (output === 'retriever') { + return vectorStore.asRetriever(k) + } else if (output === 'vectorStore') { + ;(vectorStore as any).k = k + return vectorStore + } + return vectorStore + } +} + +module.exports = { nodeClass: Valkey_VectorStores, ValkeyVectorStore, parseConnectionUrl, createGlideClient, getConnectionConfig } diff --git a/packages/components/nodes/vectorstores/Valkey/valkey.svg b/packages/components/nodes/vectorstores/Valkey/valkey.svg new file mode 100644 index 00000000000..81d8d3b7806 --- /dev/null +++ b/packages/components/nodes/vectorstores/Valkey/valkey.svg @@ -0,0 +1 @@ + \ No newline at end of file diff --git a/packages/components/package.json b/packages/components/package.json index bebf3c06f3e..17b8f28598d 100644 --- a/packages/components/package.json +++ b/packages/components/package.json @@ -115,6 +115,7 @@ "@types/jsdom": "^21.1.1", "@upstash/redis": "1.22.1", "@upstash/vector": "1.1.5", + "@valkey/valkey-glide": "2.4.0", "@zilliz/milvus2-sdk-node": "^2.2.24", "apify-client": "^2.7.1", "assemblyai": "^4.2.2", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 5cb60a2666b..89691275d41 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -536,6 +536,9 @@ importers: '@upstash/vector': specifier: 1.1.5 version: 1.1.5 + '@valkey/valkey-glide': + specifier: 2.4.0 + version: 2.4.0 '@zilliz/milvus2-sdk-node': specifier: ^2.2.24 version: 2.3.5 @@ -9639,6 +9642,44 @@ packages: '@upstash/vector@1.1.5': resolution: {integrity: sha512-55+Beu/kCwjcnzg6fnMN06v9PYU1lv9NQfQwpjrJAQTH8GOprcRsQeyXBdNHKNzoQvRnVS0ENd5CDgFoljfrAw==} + '@valkey/valkey-glide-darwin-arm64@2.4.0': + resolution: {integrity: sha512-ET1diKJc84APH5P4kisiVr8Ys2lmNV3rezMa2fLwZ1vP6pdB9KQZxbdnib3cdrqXsG2q0QAFnF39l+k56QSgeA==} + cpu: [arm64] + os: [darwin] + + '@valkey/valkey-glide-darwin-x64@2.4.0': + resolution: {integrity: sha512-hMnQTYqOugiSUH26KwVII20Pj8vacSyKqt52XIFn5y0XPwEHuHCeTdorch4dLeLT+/2h0hS0wrJBuOYGmdxRuw==} + cpu: [x64] + os: [darwin] + + '@valkey/valkey-glide-linux-arm64-gnu@2.4.0': + resolution: {integrity: sha512-8yXVEYRh1pbpWaoVWh096JrvtSjdPNVexjpAdmbyMzOuGMGnNVyjrnH8+l1iA2qBqglj+4KQPD3FMpxAGDZXXA==} + cpu: [arm64] + os: [linux] + libc: [glibc] + + '@valkey/valkey-glide-linux-arm64-musl@2.4.0': + resolution: {integrity: sha512-is8IvM2AZ2bhDXHMktOez/6owGWsdM8BsHrUT46unqCXwRWSOMUhdNG8lNxmcajhgHkM8EYCUqwQCKvhzvLW/w==} + cpu: [arm64] + os: [linux] + libc: [musl] + + '@valkey/valkey-glide-linux-x64-gnu@2.4.0': + resolution: {integrity: sha512-2CtNW0qkYLkbMULrPBCj2xb0AzBANZ0INTkXhO5HG0RhOCtXepwJQfs+PZsxJUDgODjWt6TYfsQsrkWKN9vQ2g==} + cpu: [x64] + os: [linux] + libc: [glibc] + + '@valkey/valkey-glide-linux-x64-musl@2.4.0': + resolution: {integrity: sha512-Or4z46RwjkjIwEoZGL84dvErKfbNFFT+GDsIhQ3E3pAtMLVM8s6y/eP168y13v8Z4lXNkMddFL0z3xYJDGofSg==} + cpu: [x64] + os: [linux] + libc: [musl] + + '@valkey/valkey-glide@2.4.0': + resolution: {integrity: sha512-kSD0X/TCIBoUdSrXI9HbO6ZEtAdbC5FNWNPRcJQSbIfhKpqUQkdoF+7VHhJM4u9c3OhOgrDry436KE/VAnwP2w==} + engines: {node: '>=16'} + '@vitejs/plugin-react@3.1.0': resolution: {integrity: sha512-AfgcRL8ZBhAlc3BFdigClmTUMISmmzHn7sB2h9U1odvc5U/MjWXsAaz18b/WoppUTDBzxOJwo2VdClfUcItu9g==} engines: {node: ^14.18.0 || >=16.0.0} @@ -32260,7 +32301,7 @@ snapshots: dependencies: '@types/http-errors': 2.0.4 '@types/mime': 3.0.4 - '@types/node': 20.12.12 + '@types/node': 22.16.3 '@types/shimmer@1.2.0': {} @@ -32708,6 +32749,36 @@ snapshots: '@upstash/vector@1.1.5': {} + '@valkey/valkey-glide-darwin-arm64@2.4.0': + optional: true + + '@valkey/valkey-glide-darwin-x64@2.4.0': + optional: true + + '@valkey/valkey-glide-linux-arm64-gnu@2.4.0': + optional: true + + '@valkey/valkey-glide-linux-arm64-musl@2.4.0': + optional: true + + '@valkey/valkey-glide-linux-x64-gnu@2.4.0': + optional: true + + '@valkey/valkey-glide-linux-x64-musl@2.4.0': + optional: true + + '@valkey/valkey-glide@2.4.0': + dependencies: + long: 5.3.2 + protobufjs: 7.4.0 + optionalDependencies: + '@valkey/valkey-glide-darwin-arm64': 2.4.0 + '@valkey/valkey-glide-darwin-x64': 2.4.0 + '@valkey/valkey-glide-linux-arm64-gnu': 2.4.0 + '@valkey/valkey-glide-linux-arm64-musl': 2.4.0 + '@valkey/valkey-glide-linux-x64-gnu': 2.4.0 + '@valkey/valkey-glide-linux-x64-musl': 2.4.0 + '@vitejs/plugin-react@3.1.0(vite@4.5.2(@types/node@22.16.3)(sass@1.71.1)(terser@5.29.1))': dependencies: '@babel/core': 7.24.0 From 80da960b7e3c26304e77d2e83ea599707e2dccee Mon Sep 17 00:00:00 2001 From: Riley Des Date: Tue, 2 Jun 2026 13:26:47 -0500 Subject: [PATCH 2/4] fix(valkey): improve type safety, input validation, and search options - Replace `any` in createGlideClient with GlideClientConfiguration type - Add validation for valkeyMetadataFilter after JSON.parse - Pass sortby/dialect/limit from buildQuery to GlideFt.search - Flatten array metadata values into TAG field (fixes silent drop) - Replace @ts-ignore with @ts-expect-error + explanation - Replace lodash flatten with native .flat() - Use EmbeddingsInterface instead of Embeddings class for type assertions - Import GlideClientConfiguration from @valkey/valkey-glide --- .../nodes/vectorstores/Valkey/Valkey.ts | 64 ++++++++++++------- 1 file changed, 41 insertions(+), 23 deletions(-) diff --git a/packages/components/nodes/vectorstores/Valkey/Valkey.ts b/packages/components/nodes/vectorstores/Valkey/Valkey.ts index b1cd534f1aa..a8efb73e727 100644 --- a/packages/components/nodes/vectorstores/Valkey/Valkey.ts +++ b/packages/components/nodes/vectorstores/Valkey/Valkey.ts @@ -1,9 +1,16 @@ -import { flatten } from 'lodash' import { randomUUID } from 'crypto' -import { Batch, ClusterBatch, ClusterScanCursor, GlideClient, GlideClusterClient, GlideFt, Field } from '@valkey/valkey-glide' +import { + Batch, + ClusterBatch, + ClusterScanCursor, + GlideClient, + GlideClusterClient, + GlideFt, + Field, + GlideClientConfiguration +} from '@valkey/valkey-glide' import { VectorStore } from '@langchain/core/vectorstores' import type { EmbeddingsInterface } from '@langchain/core/embeddings' -import { Embeddings } from '@langchain/core/embeddings' import { Document } from '@langchain/core/documents' import { ICommonObject, INode, INodeData, INodeOutputsValue, INodeParams, IndexingResult } from '../../../src/Interface' import { getBaseClasses, getCredentialData, getCredentialParam } from '../../../src/utils' @@ -224,8 +231,13 @@ export class ValkeyVectorStore extends VectorStore { [this.contentKey]: documents[idx]?.pageContent || '', [this.metadataKey]: JSON.stringify(metadata), [`${this.metadataKey}_tags`]: Object.values(metadata) - .filter((v) => v != null && typeof v !== 'object') - .map((v) => String(v)) + .flatMap((v) => + Array.isArray(v) + ? v.filter((x) => x != null && typeof x !== 'object').map(String) + : v != null && typeof v !== 'object' + ? [String(v)] + : [] + ) .join(',') } @@ -270,7 +282,10 @@ export class ValkeyVectorStore extends VectorStore { const searchOptions = { params: [{ key: 'vector', value: options.PARAMS.vector }], - returnFields: options.RETURN.map((field) => ({ fieldIdentifier: field })) + returnFields: options.RETURN.map((field) => ({ fieldIdentifier: field })), + sortby: options.SORTBY, + dialect: options.DIALECT, + limit: { offset: options.LIMIT.from, count: options.LIMIT.size } } const results = await GlideFt.search(this.valkeyClient, this.indexName, queryStr, searchOptions) @@ -399,19 +414,18 @@ function parseConnectionUrl(url: string): ValkeyConnectionConfig { } async function createGlideClient(config: ValkeyConnectionConfig): Promise { - const clientConfig: any = { + const clientConfig: GlideClientConfiguration = { addresses: [{ host: config.host, port: config.port }], requestTimeout: 5000, - connectionTimeout: 2000 - } - if (config.username || config.password) { - clientConfig.credentials = { - ...(config.username && { username: config.username }), - ...(config.password && { password: config.password }) - } - } - if (config.useTLS) { - clientConfig.useTLS = true + ...(config.username || config.password + ? { + credentials: { + ...(config.username && { username: config.username }), + password: config.password ?? '' + } + } + : {}), + ...(config.useTLS ? { useTLS: true } : {}) } return await GlideClient.createClient(clientConfig) } @@ -556,13 +570,13 @@ class Valkey_VectorStores implements INode { ] } - //@ts-ignore + // @ts-expect-error — upsert returns Partial; INode expects full IndexingResult | void vectorStoreMethods = { async upsert(nodeData: INodeData, options: ICommonObject): Promise> { const credentialData = await getCredentialData(nodeData.credential ?? '', options) const connectionConfig = getConnectionConfig(credentialData, nodeData) const indexName = nodeData.inputs?.indexName as string - const embeddings = nodeData.inputs?.embeddings as Embeddings + const embeddings = nodeData.inputs?.embeddings as EmbeddingsInterface const replaceIndex = nodeData.inputs?.replaceIndex as boolean const contentKey = (nodeData.inputs?.contentKey as string) || 'content' const metadataKey = (nodeData.inputs?.metadataKey as string) || 'metadata' @@ -570,7 +584,7 @@ class Valkey_VectorStores implements INode { const recordManager = nodeData.inputs?.recordManager const docs = nodeData.inputs?.document as Document[] - const flattenDocs = docs && docs.length ? flatten(docs) : [] + const flattenDocs = docs && docs.length ? docs.flat() : [] const finalDocs: Document[] = [] for (let i = 0; i < flattenDocs.length; i += 1) { if (flattenDocs[i] && flattenDocs[i].pageContent) { @@ -619,7 +633,7 @@ class Valkey_VectorStores implements INode { const credentialData = await getCredentialData(nodeData.credential ?? '', options) const connectionConfig = getConnectionConfig(credentialData, nodeData) const indexName = nodeData.inputs?.indexName as string - const embeddings = nodeData.inputs?.embeddings as Embeddings + const embeddings = nodeData.inputs?.embeddings as EmbeddingsInterface const contentKey = (nodeData.inputs?.contentKey as string) || 'content' const metadataKey = (nodeData.inputs?.metadataKey as string) || 'metadata' const vectorKey = (nodeData.inputs?.vectorKey as string) || 'content_vector' @@ -657,7 +671,7 @@ class Valkey_VectorStores implements INode { const credentialData = await getCredentialData(nodeData.credential ?? '', options) const connectionConfig = getConnectionConfig(credentialData, nodeData) const indexName = nodeData.inputs?.indexName as string - const embeddings = nodeData.inputs?.embeddings as Embeddings + const embeddings = nodeData.inputs?.embeddings as EmbeddingsInterface const topK = nodeData.inputs?.topK as string const k = topK ? Math.max(1, parseInt(topK, 10)) : 4 const output = nodeData.outputs?.output as string @@ -668,7 +682,11 @@ class Valkey_VectorStores implements INode { let filter: ValkeyVectorStoreFilterType | undefined if (valkeyMetadataFilter) { - filter = typeof valkeyMetadataFilter === 'object' ? valkeyMetadataFilter : JSON.parse(valkeyMetadataFilter) + const parsed = typeof valkeyMetadataFilter === 'object' ? valkeyMetadataFilter : JSON.parse(valkeyMetadataFilter) + if (!Array.isArray(parsed) && typeof parsed !== 'string') { + throw new Error('valkeyMetadataFilter must be a JSON array of strings or a single string') + } + filter = parsed } // Persistent client for the vector store's lifetime. Matches Postgres/TypeORM pattern in From 17cfc2e1e2fa3bf946d682f26435b099aa5d926d Mon Sep 17 00:00:00 2001 From: Riley Des Date: Tue, 2 Jun 2026 13:27:00 -0500 Subject: [PATCH 3/4] fix(valkey): correct port default type and improve test robustness - Fix port default from string '6379' to number 6379 in credential - Use cursor loop in deleteAll test scan verification - Narrow catch (err: any) to proper instanceof check --- .../credentials/ValkeyApi.credential.ts | 2 +- .../vectorstores/Valkey/Valkey.int.test.ts | 20 +++++++++++++++---- 2 files changed, 17 insertions(+), 5 deletions(-) diff --git a/packages/components/credentials/ValkeyApi.credential.ts b/packages/components/credentials/ValkeyApi.credential.ts index 9887f3396e7..158f9aaf8c8 100644 --- a/packages/components/credentials/ValkeyApi.credential.ts +++ b/packages/components/credentials/ValkeyApi.credential.ts @@ -23,7 +23,7 @@ class ValkeyApi implements INodeCredential { label: 'Port', name: 'valkeyPort', type: 'number', - default: '6379' + default: 6379 }, { label: 'User', diff --git a/packages/components/nodes/vectorstores/Valkey/Valkey.int.test.ts b/packages/components/nodes/vectorstores/Valkey/Valkey.int.test.ts index a9d21a3440a..983c4a7a0aa 100644 --- a/packages/components/nodes/vectorstores/Valkey/Valkey.int.test.ts +++ b/packages/components/nodes/vectorstores/Valkey/Valkey.int.test.ts @@ -48,13 +48,13 @@ async function checkValkeyAvailability(): Promise { requestTimeout: 3000 }) } catch (err: any) { - return `Cannot connect to Valkey at ${VALKEY_HOST}:${VALKEY_PORT}: ${err.message}` + return `Cannot connect to Valkey at ${VALKEY_HOST}:${VALKEY_PORT}: ${err instanceof Error ? err.message : String(err)}` } try { await client.customCommand(['FT._LIST']) } catch (err: any) { client.close() - return `valkey-search module not loaded at ${VALKEY_HOST}:${VALKEY_PORT}: ${err.message}` + return `valkey-search module not loaded at ${VALKEY_HOST}:${VALKEY_PORT}: ${err instanceof Error ? err.message : String(err)}` } client.close() return null @@ -168,7 +168,13 @@ describe('ValkeyVectorStore Integration', () => { ]) // Verify keys exist - const [, keysBefore] = await client.scan('0', { match: `${prefix}*`, count: 100 }) + const keysBefore: string[] = [] + let cur = '0' + do { + const [next, keys] = await client.scan(cur, { match: `${prefix}*`, count: 100 }) + cur = String(next) + keysBefore.push(...(keys as string[])) + } while (cur !== '0') expect(keysBefore.length).toBe(2) await store.delete({ deleteAll: true }) @@ -176,7 +182,13 @@ describe('ValkeyVectorStore Integration', () => { // Verify index gone expect(await store.checkIndexExists()).toBe(false) // Verify HASH keys gone - const [, keysAfter] = await client.scan('0', { match: `${prefix}*`, count: 100 }) + const keysAfter: string[] = [] + cur = '0' + do { + const [next, keys] = await client.scan(cur, { match: `${prefix}*`, count: 100 }) + cur = String(next) + keysAfter.push(...(keys as string[])) + } while (cur !== '0') expect(keysAfter.length).toBe(0) }) }) From 29998a2d11e51a1b8f065b6b73220ba123c5a79c Mon Sep 17 00:00:00 2001 From: Riley Des Date: Wed, 3 Jun 2026 09:30:01 -0500 Subject: [PATCH 4/4] fix(valkey): harden topK parsing, validate field names, remove dead code - Guard against NaN propagation when topK is non-numeric string - Validate contentKey/metadataKey/vectorKey against /^[a-zA-Z_][a-zA-Z0-9_]*$/ to prevent FT.SEARCH query injection via crafted field names - Remove unused createIndexOptions (interface, field, constructor assignment) - Document that retryStrategy is ClusterBatchOptions-only per valkey-glide API Signed-off-by: Riley Des --- .../nodes/vectorstores/Valkey/Valkey.ts | 24 ++++++++++++------- 1 file changed, 15 insertions(+), 9 deletions(-) diff --git a/packages/components/nodes/vectorstores/Valkey/Valkey.ts b/packages/components/nodes/vectorstores/Valkey/Valkey.ts index a8efb73e727..5048c6c9a6e 100644 --- a/packages/components/nodes/vectorstores/Valkey/Valkey.ts +++ b/packages/components/nodes/vectorstores/Valkey/Valkey.ts @@ -41,7 +41,6 @@ export interface ValkeyVectorStoreConfig { valkeyClient: GlideClient | GlideClusterClient indexName: string indexOptions?: CreateSchemaFlatVectorField | CreateSchemaHNSWVectorField - createIndexOptions?: Record keyPrefix?: string contentKey?: string metadataKey?: string @@ -66,8 +65,6 @@ export class ValkeyVectorStore extends VectorStore { indexOptions: CreateSchemaFlatVectorField | CreateSchemaHNSWVectorField - createIndexOptions: Record - keyPrefix: string contentKey: string @@ -96,16 +93,22 @@ export class ValkeyVectorStore extends VectorStore { this.contentKey = _dbConfig.contentKey ?? 'content' this.metadataKey = _dbConfig.metadataKey ?? 'metadata' this.vectorKey = _dbConfig.vectorKey ?? 'content_vector' + // Validate field names to prevent FT.SEARCH query injection + const fieldNameRe = /^[a-zA-Z_][a-zA-Z0-9_]*$/ + for (const [name, val] of Object.entries({ + contentKey: this.contentKey, + metadataKey: this.metadataKey, + vectorKey: this.vectorKey + })) { + if (!fieldNameRe.test(val)) { + throw new Error(`${name} must match /^[a-zA-Z_][a-zA-Z0-9_]*$/, got "${val}"`) + } + } this.filter = _dbConfig.filter if (_dbConfig.ttl !== undefined && _dbConfig.ttl <= 0) { throw new Error(`TTL must be a positive integer, got ${_dbConfig.ttl}`) } this.ttl = _dbConfig.ttl - this.createIndexOptions = { - ON: 'HASH', - PREFIX: this.keyPrefix, - ...(_dbConfig.createIndexOptions || {}) - } } async checkIndexExists(): Promise { @@ -261,6 +264,8 @@ export class ValkeyVectorStore extends VectorStore { batch.expire(key, this.ttl) } } + // Standalone BatchOptions only supports timeout; retryStrategy is ClusterBatchOptions-only. + // Standalone client handles retries internally via requestTimeout + reconnection. await (this.valkeyClient as GlideClient).exec(batch, true) } commands.length = 0 @@ -673,7 +678,8 @@ class Valkey_VectorStores implements INode { const indexName = nodeData.inputs?.indexName as string const embeddings = nodeData.inputs?.embeddings as EmbeddingsInterface const topK = nodeData.inputs?.topK as string - const k = topK ? Math.max(1, parseInt(topK, 10)) : 4 + const parsed = parseInt(topK, 10) + const k = topK && !isNaN(parsed) ? Math.max(1, parsed) : 4 const output = nodeData.outputs?.output as string const contentKey = (nodeData.inputs?.contentKey as string) || 'content' const metadataKey = (nodeData.inputs?.metadataKey as string) || 'metadata'