926 lines
29 KiB
TypeScript
926 lines
29 KiB
TypeScript
import {
|
|
type AddFactInput,
|
|
type ConnectedTopic,
|
|
type Fact,
|
|
type FactTopic,
|
|
type FindSimilarFactsInput,
|
|
type IndexFactEmbeddingsInput,
|
|
type LinkTopicsInput,
|
|
type ListTopicsOptions,
|
|
type ScoredFact,
|
|
type SearchFactsInput,
|
|
type Space,
|
|
type SpaceScopedInput,
|
|
type Topic,
|
|
type TopicLookupOptions,
|
|
type TopicWithFacts,
|
|
type UpsertSpaceInput,
|
|
type UpsertTopicInput,
|
|
} from '../types/api';
|
|
import type { IngestStatementOptions } from '../ingestion/types';
|
|
import type { DatabaseConnection, IdentityDBConnectionConfig } from '../adapters/dialect';
|
|
import type { IdentityDatabaseSchema } from '../types/database';
|
|
import type { FactRecord, SpaceRecord, TopicRecord } from '../types/domain';
|
|
import { createDatabase } from '../adapters/dialect';
|
|
import { extractFact } from '../ingestion/extractor';
|
|
import {
|
|
findFactRowsConnectingTopicIds,
|
|
findFactRowsForTopicId,
|
|
findTopicLinksForFactIds,
|
|
} from '../queries/facts';
|
|
import {
|
|
type DatabaseExecutor,
|
|
findChildTopicRows,
|
|
findConnectedTopicRows,
|
|
findParentTopicRows,
|
|
findSpaceRowByNormalizedName,
|
|
findTopicRowByNameOrAlias,
|
|
findTopicRowByNormalizedAlias,
|
|
findTopicRowByNormalizedName,
|
|
listTopicAliasRowsForTopicId,
|
|
listTopicRows,
|
|
} from '../queries/topics';
|
|
import { IdentityDBError } from './errors';
|
|
import { initializeSchema } from './migrations';
|
|
import {
|
|
canonicalizeSpaceName,
|
|
canonicalizeTopicName,
|
|
cosineSimilarity,
|
|
createContentHash,
|
|
createId,
|
|
deserializeEmbedding,
|
|
mapFactRow,
|
|
mapSpaceRow,
|
|
mapTopicRow,
|
|
normalizeSpaceName,
|
|
normalizeTopicName,
|
|
nowIsoString,
|
|
serializeEmbedding,
|
|
serializeMetadata,
|
|
} from './utils';
|
|
|
|
const DEFAULT_SPACE_NAME = 'default';
|
|
|
|
export class IdentityDB {
|
|
private constructor(private readonly connection: DatabaseConnection) {}
|
|
|
|
static async connect(config: IdentityDBConnectionConfig): Promise<IdentityDB> {
|
|
const connection = await createDatabase(config);
|
|
return new IdentityDB(connection);
|
|
}
|
|
|
|
async initialize(): Promise<void> {
|
|
await initializeSchema(this.connection.db);
|
|
}
|
|
|
|
async close(): Promise<void> {
|
|
await this.connection.destroy();
|
|
}
|
|
|
|
async upsertSpace(input: UpsertSpaceInput): Promise<Space> {
|
|
return this.connection.db.transaction().execute(async (trx) => {
|
|
const normalizedName = normalizeSpaceName(input.name);
|
|
if (normalizedName.length === 0) {
|
|
throw new IdentityDBError('Space name cannot be empty.');
|
|
}
|
|
|
|
const now = nowIsoString();
|
|
const existing = await findSpaceRowByNormalizedName(trx, normalizedName);
|
|
|
|
if (existing) {
|
|
await trx
|
|
.updateTable('spaces')
|
|
.set({
|
|
name: canonicalizeSpaceName(input.name),
|
|
description: input.description !== undefined ? input.description : existing.description,
|
|
metadata: input.metadata !== undefined ? serializeMetadata(input.metadata) : existing.metadata,
|
|
updated_at: now,
|
|
})
|
|
.where('id', '=', existing.id)
|
|
.execute();
|
|
|
|
const updated = await trx
|
|
.selectFrom('spaces')
|
|
.selectAll()
|
|
.where('id', '=', existing.id)
|
|
.executeTakeFirstOrThrow();
|
|
|
|
return mapSpaceRow(updated);
|
|
}
|
|
|
|
const createdRow: SpaceRecord = {
|
|
id: createId(),
|
|
name: canonicalizeSpaceName(input.name),
|
|
normalized_name: normalizedName,
|
|
description: input.description ?? null,
|
|
metadata: serializeMetadata(input.metadata),
|
|
created_at: now,
|
|
updated_at: now,
|
|
};
|
|
|
|
await trx.insertInto('spaces').values(createdRow).execute();
|
|
return mapSpaceRow(createdRow);
|
|
});
|
|
}
|
|
|
|
async getSpaceByName(name: string): Promise<Space | null> {
|
|
const normalizedName = normalizeSpaceName(name);
|
|
if (normalizedName.length === 0) {
|
|
return null;
|
|
}
|
|
|
|
const row = await findSpaceRowByNormalizedName(this.connection.db, normalizedName);
|
|
return row ? mapSpaceRow(row) : null;
|
|
}
|
|
|
|
async listSpaces(): Promise<Space[]> {
|
|
const rows = await this.connection.db
|
|
.selectFrom('spaces')
|
|
.selectAll()
|
|
.orderBy('normalized_name', 'asc')
|
|
.execute();
|
|
|
|
return rows.map(mapSpaceRow);
|
|
}
|
|
|
|
async upsertTopic(input: UpsertTopicInput): Promise<Topic> {
|
|
return this.upsertTopicInExecutor(this.connection.db, input);
|
|
}
|
|
|
|
async addFact(input: AddFactInput): Promise<Fact> {
|
|
if (input.statement.trim().length === 0) {
|
|
throw new IdentityDBError('Fact statement cannot be empty.');
|
|
}
|
|
|
|
if (input.topics.length === 0) {
|
|
throw new IdentityDBError('A fact must be linked to at least one topic.');
|
|
}
|
|
|
|
return this.connection.db.transaction().execute(async (trx) => {
|
|
const space = await this.getOrCreateSpaceInExecutor(trx, input.spaceName);
|
|
const createdAt = nowIsoString();
|
|
const factId = createId();
|
|
|
|
await trx
|
|
.insertInto('facts')
|
|
.values({
|
|
id: factId,
|
|
space_id: space.id,
|
|
statement: input.statement.trim(),
|
|
summary: input.summary ?? null,
|
|
source: input.source ?? null,
|
|
confidence: input.confidence ?? null,
|
|
metadata: serializeMetadata(input.metadata),
|
|
created_at: createdAt,
|
|
updated_at: createdAt,
|
|
})
|
|
.execute();
|
|
|
|
const topics: FactTopic[] = [];
|
|
|
|
for (let index = 0; index < input.topics.length; index += 1) {
|
|
const topicInput = input.topics[index]!;
|
|
this.assertScopedTopicInput(space, topicInput.spaceName);
|
|
const topic = await this.upsertTopicInExecutor(trx, {
|
|
...topicInput,
|
|
spaceName: space.name,
|
|
});
|
|
|
|
await trx
|
|
.insertInto('fact_topics')
|
|
.values({
|
|
fact_id: factId,
|
|
topic_id: topic.id,
|
|
role: topicInput.role ?? null,
|
|
position: index,
|
|
created_at: createdAt,
|
|
})
|
|
.execute();
|
|
|
|
topics.push({
|
|
...topic,
|
|
role: topicInput.role ?? null,
|
|
position: index,
|
|
});
|
|
}
|
|
|
|
return {
|
|
id: factId,
|
|
spaceId: space.id,
|
|
statement: input.statement.trim(),
|
|
summary: input.summary ?? null,
|
|
source: input.source ?? null,
|
|
confidence: input.confidence ?? null,
|
|
metadata: input.metadata ?? null,
|
|
createdAt,
|
|
updatedAt: createdAt,
|
|
topics,
|
|
};
|
|
});
|
|
}
|
|
|
|
async ingestStatement(statement: string, options: IngestStatementOptions): Promise<Fact> {
|
|
const extracted = await extractFact(statement, options.extractor);
|
|
const factInput: AddFactInput = {
|
|
statement: extracted.statement ?? statement,
|
|
topics: extracted.topics,
|
|
spaceName: options.spaceName,
|
|
};
|
|
|
|
if (extracted.summary !== undefined) {
|
|
factInput.summary = extracted.summary;
|
|
}
|
|
|
|
if (extracted.source !== undefined) {
|
|
factInput.source = extracted.source;
|
|
}
|
|
|
|
if (extracted.confidence !== undefined) {
|
|
factInput.confidence = extracted.confidence;
|
|
}
|
|
|
|
if (extracted.metadata !== undefined) {
|
|
factInput.metadata = extracted.metadata;
|
|
}
|
|
|
|
if (options.embeddingProvider) {
|
|
const similarFacts = await this.findSimilarFacts({
|
|
statement: factInput.statement,
|
|
provider: options.embeddingProvider,
|
|
topicNames: factInput.topics.map((topic) => topic.name),
|
|
limit: 1,
|
|
minimumScore: options.duplicateThreshold ?? 0.97,
|
|
spaceName: options.spaceName,
|
|
});
|
|
|
|
if (similarFacts[0]) {
|
|
return similarFacts[0];
|
|
}
|
|
}
|
|
|
|
const fact = await this.addFact(factInput);
|
|
|
|
if (options.embeddingProvider) {
|
|
await this.indexFactEmbedding(fact.id, {
|
|
provider: options.embeddingProvider,
|
|
spaceName: options.spaceName,
|
|
});
|
|
}
|
|
|
|
return fact;
|
|
}
|
|
|
|
async indexFactEmbeddings(input: IndexFactEmbeddingsInput): Promise<void> {
|
|
const space = await this.getSpaceForRead(input.spaceName);
|
|
if (input.spaceName && !space) {
|
|
return;
|
|
}
|
|
|
|
let factQuery = this.connection.db.selectFrom('facts').selectAll().orderBy('created_at', 'asc');
|
|
if (space) {
|
|
factQuery = factQuery.where('space_id', '=', space.id);
|
|
}
|
|
|
|
const factRows = await factQuery.execute();
|
|
if (factRows.length === 0) {
|
|
return;
|
|
}
|
|
|
|
const embeddings = input.provider.embedMany
|
|
? await input.provider.embedMany(factRows.map((factRow) => factRow.statement))
|
|
: await Promise.all(factRows.map((factRow) => input.provider.embed(factRow.statement)));
|
|
|
|
if (embeddings.length !== factRows.length) {
|
|
throw new IdentityDBError('Embedding provider returned a mismatched number of embeddings.');
|
|
}
|
|
|
|
await this.connection.db.transaction().execute(async (trx) => {
|
|
for (let index = 0; index < factRows.length; index += 1) {
|
|
const factRow = factRows[index]!;
|
|
const embedding = embeddings[index]!;
|
|
this.assertEmbeddingShape(embedding, input.provider.dimensions);
|
|
await this.upsertFactEmbeddingRecord(trx, factRow.id, factRow.statement, embedding, input.provider.model);
|
|
}
|
|
});
|
|
}
|
|
|
|
async indexFactEmbedding(factId: string, input: IndexFactEmbeddingsInput): Promise<void> {
|
|
const factRow = await this.connection.db
|
|
.selectFrom('facts')
|
|
.selectAll()
|
|
.where('id', '=', factId)
|
|
.executeTakeFirst();
|
|
|
|
if (!factRow) {
|
|
throw new IdentityDBError(`Fact not found: ${factId}`);
|
|
}
|
|
|
|
if (input.spaceName) {
|
|
const space = await this.getSpaceForRead(input.spaceName);
|
|
if (!space || space.id !== factRow.space_id) {
|
|
throw new IdentityDBError(`Fact ${factId} does not belong to space ${canonicalizeSpaceName(input.spaceName)}.`);
|
|
}
|
|
}
|
|
|
|
const embedding = await input.provider.embed(factRow.statement);
|
|
this.assertEmbeddingShape(embedding, input.provider.dimensions);
|
|
|
|
await this.connection.db.transaction().execute(async (trx) => {
|
|
await this.upsertFactEmbeddingRecord(trx, factRow.id, factRow.statement, embedding, input.provider.model);
|
|
});
|
|
}
|
|
|
|
async searchFacts(input: SearchFactsInput): Promise<ScoredFact[]> {
|
|
const queryText = input.query.trim();
|
|
if (queryText.length === 0) {
|
|
return [];
|
|
}
|
|
|
|
const space = await this.getSpaceForRead(input.spaceName);
|
|
if (input.spaceName && !space) {
|
|
return [];
|
|
}
|
|
|
|
const queryEmbedding = await input.provider.embed(queryText);
|
|
this.assertEmbeddingShape(queryEmbedding, input.provider.dimensions);
|
|
|
|
return this.searchFactsByEmbedding({
|
|
providerModel: input.provider.model,
|
|
queryEmbedding,
|
|
topicNames: input.topicNames,
|
|
limit: input.limit,
|
|
minimumScore: input.minimumScore,
|
|
spaceId: space?.id,
|
|
});
|
|
}
|
|
|
|
async findSimilarFacts(input: FindSimilarFactsInput): Promise<ScoredFact[]> {
|
|
const statement = input.statement.trim();
|
|
if (statement.length === 0) {
|
|
return [];
|
|
}
|
|
|
|
const space = await this.getSpaceForRead(input.spaceName);
|
|
if (input.spaceName && !space) {
|
|
return [];
|
|
}
|
|
|
|
const queryEmbedding = await input.provider.embed(statement);
|
|
this.assertEmbeddingShape(queryEmbedding, input.provider.dimensions);
|
|
|
|
return this.searchFactsByEmbedding({
|
|
providerModel: input.provider.model,
|
|
queryEmbedding,
|
|
topicNames: input.topicNames,
|
|
limit: input.limit,
|
|
minimumScore: input.minimumScore,
|
|
spaceId: space?.id,
|
|
});
|
|
}
|
|
|
|
async linkTopics(input: LinkTopicsInput): Promise<void> {
|
|
const parentNormalizedName = normalizeTopicName(input.parentName);
|
|
const childNormalizedName = normalizeTopicName(input.childName);
|
|
|
|
if (parentNormalizedName.length === 0 || childNormalizedName.length === 0) {
|
|
throw new IdentityDBError('Topic hierarchy links require both a parent and child topic name.');
|
|
}
|
|
|
|
if (parentNormalizedName === childNormalizedName) {
|
|
throw new IdentityDBError('A topic cannot be linked as its own parent.');
|
|
}
|
|
|
|
await this.connection.db.transaction().execute(async (trx) => {
|
|
const space = await this.getOrCreateSpaceInExecutor(trx, input.spaceName);
|
|
const parentTopic = await this.upsertTopicInExecutor(trx, {
|
|
name: input.parentName,
|
|
granularity: 'abstract',
|
|
spaceName: space.name,
|
|
});
|
|
const childTopic = await this.upsertTopicInExecutor(trx, {
|
|
name: input.childName,
|
|
spaceName: space.name,
|
|
});
|
|
|
|
const existing = await trx
|
|
.selectFrom('topic_relations')
|
|
.select(['parent_topic_id'])
|
|
.where('parent_topic_id', '=', parentTopic.id)
|
|
.where('child_topic_id', '=', childTopic.id)
|
|
.where('relation', '=', 'parent_of')
|
|
.executeTakeFirst();
|
|
|
|
if (!existing) {
|
|
await trx
|
|
.insertInto('topic_relations')
|
|
.values({
|
|
parent_topic_id: parentTopic.id,
|
|
child_topic_id: childTopic.id,
|
|
relation: 'parent_of',
|
|
created_at: nowIsoString(),
|
|
})
|
|
.execute();
|
|
}
|
|
});
|
|
}
|
|
|
|
async addTopicAlias(canonicalName: string, alias: string, options?: SpaceScopedInput): Promise<void> {
|
|
const normalizedAlias = normalizeTopicName(alias);
|
|
|
|
if (normalizedAlias.length === 0) {
|
|
throw new IdentityDBError('Topic alias cannot be empty.');
|
|
}
|
|
|
|
await this.connection.db.transaction().execute(async (trx) => {
|
|
const space = await this.getOrCreateSpaceInExecutor(trx, options?.spaceName);
|
|
const canonicalTopic = await this.upsertTopicInExecutor(trx, {
|
|
name: canonicalName,
|
|
spaceName: space.name,
|
|
});
|
|
|
|
if (normalizedAlias === canonicalTopic.normalizedName) {
|
|
return;
|
|
}
|
|
|
|
const exactTopicMatch = await findTopicRowByNormalizedName(trx, space.id, normalizedAlias);
|
|
if (exactTopicMatch && exactTopicMatch.id !== canonicalTopic.id) {
|
|
throw new IdentityDBError('Cannot assign an alias that already belongs to another canonical topic.');
|
|
}
|
|
|
|
const aliasMatch = await findTopicRowByNormalizedAlias(trx, space.id, normalizedAlias);
|
|
if (aliasMatch) {
|
|
if (aliasMatch.id !== canonicalTopic.id) {
|
|
throw new IdentityDBError('Cannot assign an alias that already resolves to another topic.');
|
|
}
|
|
return;
|
|
}
|
|
|
|
const createdAt = nowIsoString();
|
|
await trx
|
|
.insertInto('topic_aliases')
|
|
.values({
|
|
id: createId(),
|
|
space_id: space.id,
|
|
topic_id: canonicalTopic.id,
|
|
alias: canonicalizeTopicName(alias),
|
|
normalized_alias: normalizedAlias,
|
|
is_primary: 0,
|
|
created_at: createdAt,
|
|
updated_at: createdAt,
|
|
})
|
|
.execute();
|
|
});
|
|
}
|
|
|
|
async resolveTopic(name: string, options?: SpaceScopedInput): Promise<Topic | null> {
|
|
const topicRow = await this.getRequiredTopicRow(name, options?.spaceName);
|
|
return topicRow ? mapTopicRow(topicRow) : null;
|
|
}
|
|
|
|
async getTopicAliases(name: string, options?: SpaceScopedInput): Promise<string[]> {
|
|
const topicRow = await this.getRequiredTopicRow(name, options?.spaceName);
|
|
if (!topicRow) {
|
|
return [];
|
|
}
|
|
|
|
const aliasRows = await listTopicAliasRowsForTopicId(this.connection.db, topicRow.space_id, topicRow.id);
|
|
return aliasRows.map((aliasRow) => aliasRow.alias);
|
|
}
|
|
|
|
async getTopicChildren(name: string, options?: SpaceScopedInput): Promise<Topic[]> {
|
|
const topicRow = await this.getRequiredTopicRow(name, options?.spaceName);
|
|
if (!topicRow) {
|
|
return [];
|
|
}
|
|
|
|
const childRows = await findChildTopicRows(this.connection.db, topicRow.space_id, topicRow.id);
|
|
return childRows.map(mapTopicRow);
|
|
}
|
|
|
|
async getTopicParents(name: string, options?: SpaceScopedInput): Promise<Topic[]> {
|
|
const topicRow = await this.getRequiredTopicRow(name, options?.spaceName);
|
|
if (!topicRow) {
|
|
return [];
|
|
}
|
|
|
|
const parentRows = await findParentTopicRows(this.connection.db, topicRow.space_id, topicRow.id);
|
|
return parentRows.map(mapTopicRow);
|
|
}
|
|
|
|
async getTopicLineage(name: string, options?: SpaceScopedInput): Promise<Topic[]> {
|
|
const topicRow = await this.getRequiredTopicRow(name, options?.spaceName);
|
|
if (!topicRow) {
|
|
return [];
|
|
}
|
|
|
|
const lineage: Topic[] = [];
|
|
const visitedTopicIds = new Set<string>([topicRow.id]);
|
|
let currentLevelIds = [topicRow.id];
|
|
|
|
while (currentLevelIds.length > 0) {
|
|
const nextLevelIds: string[] = [];
|
|
|
|
for (const currentId of currentLevelIds) {
|
|
const parentRows = await findParentTopicRows(this.connection.db, topicRow.space_id, currentId);
|
|
for (const parentRow of parentRows) {
|
|
if (visitedTopicIds.has(parentRow.id)) {
|
|
continue;
|
|
}
|
|
|
|
visitedTopicIds.add(parentRow.id);
|
|
nextLevelIds.push(parentRow.id);
|
|
lineage.push(mapTopicRow(parentRow));
|
|
}
|
|
}
|
|
|
|
currentLevelIds = nextLevelIds;
|
|
}
|
|
|
|
return lineage;
|
|
}
|
|
|
|
async getTopicFacts(name: string, options?: SpaceScopedInput): Promise<Fact[]> {
|
|
const topicRow = await this.getRequiredTopicRow(name, options?.spaceName);
|
|
if (!topicRow) {
|
|
return [];
|
|
}
|
|
|
|
const factRows = await findFactRowsForTopicId(this.connection.db, topicRow.space_id, topicRow.id);
|
|
return this.hydrateFacts(factRows, topicRow.space_id);
|
|
}
|
|
|
|
async getTopicFactsLinkedTo(name: string, linkedTopicName: string, options?: SpaceScopedInput): Promise<Fact[]> {
|
|
return this.findFactsConnectingTopics([name, linkedTopicName], options);
|
|
}
|
|
|
|
async findFactsConnectingTopics(names: string[], options?: SpaceScopedInput): Promise<Fact[]> {
|
|
if (names.length === 0) {
|
|
return [];
|
|
}
|
|
|
|
const space = await this.getSpaceForRead(options?.spaceName);
|
|
if (options?.spaceName && !space) {
|
|
return [];
|
|
}
|
|
|
|
const topicRows = await Promise.all(names.map((name) => this.getRequiredTopicRow(name, options?.spaceName)));
|
|
if (topicRows.some((topicRow) => topicRow === undefined)) {
|
|
return [];
|
|
}
|
|
|
|
const topicIds = topicRows.map((topicRow) => topicRow!.id);
|
|
const spaceId = topicRows[0]!.space_id ?? space?.id;
|
|
const factRows = await findFactRowsConnectingTopicIds(this.connection.db, spaceId, topicIds);
|
|
|
|
return this.hydrateFacts(factRows, spaceId);
|
|
}
|
|
|
|
async getTopicByName(name: string, options: { includeFacts: true; spaceName?: string }): Promise<TopicWithFacts | null>;
|
|
async getTopicByName(name: string, options?: TopicLookupOptions): Promise<Topic | null>;
|
|
async getTopicByName(name: string, options?: TopicLookupOptions): Promise<Topic | TopicWithFacts | null> {
|
|
const topicRow = await this.getRequiredTopicRow(name, options?.spaceName);
|
|
if (!topicRow) {
|
|
return null;
|
|
}
|
|
|
|
const topic = mapTopicRow(topicRow);
|
|
if (options?.includeFacts) {
|
|
return {
|
|
...topic,
|
|
facts: await this.getTopicFacts(name, { spaceName: options.spaceName }),
|
|
};
|
|
}
|
|
|
|
return topic;
|
|
}
|
|
|
|
async listTopics(options: { includeFacts: true; limit?: number; spaceName?: string }): Promise<TopicWithFacts[]>;
|
|
async listTopics(options?: ListTopicsOptions): Promise<Topic[]>;
|
|
async listTopics(options?: ListTopicsOptions): Promise<Topic[] | TopicWithFacts[]> {
|
|
const space = await this.getSpaceForRead(options?.spaceName);
|
|
if (options?.spaceName && !space) {
|
|
return [];
|
|
}
|
|
|
|
const spaceId = space?.id ?? await this.getDefaultSpaceIdForRead();
|
|
if (!spaceId) {
|
|
return [];
|
|
}
|
|
|
|
const rows = await listTopicRows(this.connection.db, spaceId, options?.limit);
|
|
if (!options?.includeFacts) {
|
|
return rows.map(mapTopicRow);
|
|
}
|
|
|
|
const topicsWithFacts: TopicWithFacts[] = [];
|
|
for (const row of rows) {
|
|
topicsWithFacts.push({
|
|
...mapTopicRow(row),
|
|
facts: await this.getTopicFacts(row.name, { spaceName: options?.spaceName }),
|
|
});
|
|
}
|
|
|
|
return topicsWithFacts;
|
|
}
|
|
|
|
async findConnectedTopics(name: string, options?: SpaceScopedInput): Promise<ConnectedTopic[]> {
|
|
const topicRow = await this.getRequiredTopicRow(name, options?.spaceName);
|
|
if (!topicRow) {
|
|
return [];
|
|
}
|
|
|
|
const rows = await findConnectedTopicRows(this.connection.db, topicRow.space_id, topicRow.id);
|
|
return rows.map((row) => ({
|
|
...mapTopicRow(row),
|
|
sharedFactCount: row.shared_fact_count,
|
|
}));
|
|
}
|
|
|
|
private async searchFactsByEmbedding(input: {
|
|
providerModel: string;
|
|
queryEmbedding: number[];
|
|
topicNames?: string[] | undefined;
|
|
limit?: number | undefined;
|
|
minimumScore?: number | undefined;
|
|
spaceId?: string | undefined;
|
|
}): Promise<ScoredFact[]> {
|
|
const effectiveSpaceId = input.spaceId ?? await this.getDefaultSpaceIdForRead();
|
|
if (!effectiveSpaceId) {
|
|
return [];
|
|
}
|
|
|
|
const topicIds = await this.resolveTopicIds(input.topicNames, effectiveSpaceId);
|
|
if (topicIds === null) {
|
|
return [];
|
|
}
|
|
|
|
const factRows = topicIds.length > 0
|
|
? await findFactRowsConnectingTopicIds(this.connection.db, effectiveSpaceId, topicIds)
|
|
: await this.connection.db
|
|
.selectFrom('facts')
|
|
.innerJoin('fact_embeddings', 'fact_embeddings.fact_id', 'facts.id')
|
|
.selectAll('facts')
|
|
.where('facts.space_id', '=', effectiveSpaceId)
|
|
.where('fact_embeddings.model', '=', input.providerModel)
|
|
.orderBy('facts.created_at', 'asc')
|
|
.execute();
|
|
|
|
if (factRows.length === 0) {
|
|
return [];
|
|
}
|
|
|
|
const embeddingRows = await this.connection.db
|
|
.selectFrom('fact_embeddings')
|
|
.innerJoin('facts', 'facts.id', 'fact_embeddings.fact_id')
|
|
.selectAll('fact_embeddings')
|
|
.where('facts.space_id', '=', effectiveSpaceId)
|
|
.where('fact_embeddings.model', '=', input.providerModel)
|
|
.where('fact_embeddings.fact_id', 'in', factRows.map((factRow) => factRow.id))
|
|
.execute();
|
|
|
|
const embeddingsByFactId = new Map(
|
|
embeddingRows.map((embeddingRow) => [embeddingRow.fact_id, deserializeEmbedding(embeddingRow.embedding)]),
|
|
);
|
|
|
|
const scoredRows = factRows
|
|
.map((factRow) => ({
|
|
factRow,
|
|
score: cosineSimilarity(input.queryEmbedding, embeddingsByFactId.get(factRow.id) ?? []),
|
|
}))
|
|
.filter((entry) => entry.score >= (input.minimumScore ?? 0))
|
|
.sort((left, right) => {
|
|
if (right.score !== left.score) {
|
|
return right.score - left.score;
|
|
}
|
|
return left.factRow.created_at.localeCompare(right.factRow.created_at);
|
|
})
|
|
.slice(0, input.limit ?? 5);
|
|
|
|
if (scoredRows.length === 0) {
|
|
return [];
|
|
}
|
|
|
|
const hydratedFacts = await this.hydrateFacts(scoredRows.map((entry) => entry.factRow), effectiveSpaceId);
|
|
const factsById = new Map(hydratedFacts.map((fact) => [fact.id, fact]));
|
|
|
|
return scoredRows.map((entry) => ({
|
|
...factsById.get(entry.factRow.id)!,
|
|
score: entry.score,
|
|
}));
|
|
}
|
|
|
|
private async resolveTopicIds(topicNames: string[] | undefined, spaceId: string): Promise<string[] | null> {
|
|
if (!topicNames || topicNames.length === 0) {
|
|
return [];
|
|
}
|
|
|
|
const topicRows = await Promise.all(topicNames.map((topicName) => this.getRequiredTopicRowInSpaceId(topicName, spaceId)));
|
|
if (topicRows.some((topicRow) => !topicRow)) {
|
|
return null;
|
|
}
|
|
|
|
return topicRows.map((topicRow) => topicRow!.id);
|
|
}
|
|
|
|
private async upsertFactEmbeddingRecord(
|
|
executor: DatabaseExecutor,
|
|
factId: string,
|
|
statement: string,
|
|
embedding: number[],
|
|
model: string,
|
|
): Promise<void> {
|
|
const timestamp = nowIsoString();
|
|
|
|
await executor
|
|
.deleteFrom('fact_embeddings')
|
|
.where('fact_id', '=', factId)
|
|
.where('model', '=', model)
|
|
.execute();
|
|
|
|
await executor
|
|
.insertInto('fact_embeddings')
|
|
.values({
|
|
fact_id: factId,
|
|
model,
|
|
dimensions: embedding.length,
|
|
embedding: serializeEmbedding(embedding),
|
|
content_hash: createContentHash(statement),
|
|
created_at: timestamp,
|
|
updated_at: timestamp,
|
|
})
|
|
.execute();
|
|
}
|
|
|
|
private assertEmbeddingShape(embedding: number[], expectedDimensions: number): void {
|
|
if (embedding.length !== expectedDimensions) {
|
|
throw new IdentityDBError(
|
|
`Embedding dimension mismatch. Expected ${expectedDimensions}, received ${embedding.length}.`,
|
|
);
|
|
}
|
|
}
|
|
|
|
private async upsertTopicInExecutor(executor: DatabaseExecutor, input: UpsertTopicInput): Promise<Topic> {
|
|
const normalizedName = normalizeTopicName(input.name);
|
|
if (normalizedName.length === 0) {
|
|
throw new IdentityDBError('Topic name cannot be empty.');
|
|
}
|
|
|
|
const space = await this.getOrCreateSpaceInExecutor(executor, input.spaceName);
|
|
const existing = await findTopicRowByNormalizedName(executor, space.id, normalizedName);
|
|
const now = nowIsoString();
|
|
|
|
if (existing) {
|
|
return this.updateTopicRowInExecutor(executor, existing, input, now, true);
|
|
}
|
|
|
|
const aliasedTopic = await findTopicRowByNormalizedAlias(executor, space.id, normalizedName);
|
|
if (aliasedTopic) {
|
|
return this.updateTopicRowInExecutor(executor, aliasedTopic, input, now, false);
|
|
}
|
|
|
|
const createdRow: TopicRecord = {
|
|
id: createId(),
|
|
space_id: space.id,
|
|
name: canonicalizeTopicName(input.name),
|
|
normalized_name: normalizedName,
|
|
category: input.category ?? 'custom',
|
|
granularity: input.granularity ?? 'mixed',
|
|
description: input.description ?? null,
|
|
metadata: serializeMetadata(input.metadata),
|
|
created_at: now,
|
|
updated_at: now,
|
|
};
|
|
|
|
await executor.insertInto('topics').values(createdRow).execute();
|
|
return mapTopicRow(createdRow);
|
|
}
|
|
|
|
private async updateTopicRowInExecutor(
|
|
executor: DatabaseExecutor,
|
|
existing: TopicRecord,
|
|
input: UpsertTopicInput,
|
|
now: string,
|
|
shouldRename: boolean,
|
|
): Promise<Topic> {
|
|
await executor
|
|
.updateTable('topics')
|
|
.set({
|
|
name: shouldRename ? canonicalizeTopicName(input.name) : existing.name,
|
|
category: input.category ?? existing.category,
|
|
granularity: input.granularity ?? existing.granularity,
|
|
description: input.description !== undefined ? input.description : existing.description,
|
|
metadata: input.metadata !== undefined ? serializeMetadata(input.metadata) : existing.metadata,
|
|
updated_at: now,
|
|
})
|
|
.where('id', '=', existing.id)
|
|
.execute();
|
|
|
|
const updated = await executor
|
|
.selectFrom('topics')
|
|
.selectAll()
|
|
.where('id', '=', existing.id)
|
|
.executeTakeFirstOrThrow();
|
|
|
|
return mapTopicRow(updated);
|
|
}
|
|
|
|
private async getRequiredTopicRow(name: string, spaceName?: string): Promise<TopicRecord | undefined> {
|
|
const space = await this.getSpaceForRead(spaceName);
|
|
if (spaceName && !space) {
|
|
return undefined;
|
|
}
|
|
|
|
const spaceId = space?.id ?? await this.getDefaultSpaceIdForRead();
|
|
if (!spaceId) {
|
|
return undefined;
|
|
}
|
|
|
|
return this.getRequiredTopicRowInSpaceId(name, spaceId);
|
|
}
|
|
|
|
private async getRequiredTopicRowInSpaceId(name: string, spaceId: string): Promise<TopicRecord | undefined> {
|
|
const normalizedName = normalizeTopicName(name);
|
|
if (normalizedName.length === 0) {
|
|
return undefined;
|
|
}
|
|
|
|
return findTopicRowByNameOrAlias(this.connection.db, spaceId, normalizedName);
|
|
}
|
|
|
|
private async hydrateFacts(factRows: FactRecord[], spaceId?: string): Promise<Fact[]> {
|
|
if (factRows.length === 0) {
|
|
return [];
|
|
}
|
|
|
|
const effectiveSpaceId = spaceId ?? factRows[0]!.space_id;
|
|
const factIds = factRows.map((fact) => fact.id);
|
|
const topicLinks = await findTopicLinksForFactIds(this.connection.db, effectiveSpaceId, factIds);
|
|
|
|
const topicsByFactId = new Map<string, FactTopic[]>();
|
|
for (const topicLink of topicLinks) {
|
|
const topics = topicsByFactId.get(topicLink.fact_id) ?? [];
|
|
topics.push({
|
|
...mapTopicRow(topicLink),
|
|
role: topicLink.role,
|
|
position: topicLink.position,
|
|
});
|
|
topicsByFactId.set(topicLink.fact_id, topics);
|
|
}
|
|
|
|
return factRows.map((factRow) => mapFactRow(factRow, topicsByFactId.get(factRow.id) ?? []));
|
|
}
|
|
|
|
private async getOrCreateSpaceInExecutor(executor: DatabaseExecutor, requestedSpaceName?: string): Promise<SpaceRecord> {
|
|
const normalizedName = normalizeSpaceName(requestedSpaceName ?? DEFAULT_SPACE_NAME);
|
|
const canonicalName = canonicalizeSpaceName(requestedSpaceName ?? DEFAULT_SPACE_NAME);
|
|
const existing = await findSpaceRowByNormalizedName(executor, normalizedName);
|
|
if (existing) {
|
|
return existing;
|
|
}
|
|
|
|
const now = nowIsoString();
|
|
const createdRow: SpaceRecord = {
|
|
id: createId(),
|
|
name: canonicalName,
|
|
normalized_name: normalizedName,
|
|
description: null,
|
|
metadata: null,
|
|
created_at: now,
|
|
updated_at: now,
|
|
};
|
|
|
|
await executor.insertInto('spaces').values(createdRow).execute();
|
|
return createdRow;
|
|
}
|
|
|
|
private async getSpaceForRead(spaceName?: string): Promise<SpaceRecord | undefined> {
|
|
if (!spaceName) {
|
|
return undefined;
|
|
}
|
|
|
|
const normalizedName = normalizeSpaceName(spaceName);
|
|
if (normalizedName.length === 0) {
|
|
return undefined;
|
|
}
|
|
|
|
return findSpaceRowByNormalizedName(this.connection.db, normalizedName);
|
|
}
|
|
|
|
private async getDefaultSpaceIdForRead(): Promise<string | undefined> {
|
|
const defaultSpace = await findSpaceRowByNormalizedName(this.connection.db, normalizeSpaceName(DEFAULT_SPACE_NAME));
|
|
return defaultSpace?.id;
|
|
}
|
|
|
|
private assertScopedTopicInput(space: SpaceRecord, topicSpaceName?: string): void {
|
|
if (!topicSpaceName) {
|
|
return;
|
|
}
|
|
|
|
if (normalizeSpaceName(topicSpaceName) !== space.normalized_name) {
|
|
throw new IdentityDBError(
|
|
`Fact topics cannot point to a different space than the fact itself (${space.name}).`,
|
|
);
|
|
}
|
|
}
|
|
}
|