This repository has been archived on 2026-04-03. You can view files and clone it. You cannot open issues or pull requests or push a commit.
Files
core/server/store/sqlite/knowledge_bases.go
2026-03-19 18:50:27 +00:00

628 lines
19 KiB
Go

package sqlite
import (
"context"
"database/sql"
"encoding/json"
"fmt"
"math"
"sort"
"strings"
"time"
"chat-switchboard/models"
"chat-switchboard/store"
)
// ── KnowledgeBaseStore ──────────────────────────
type KnowledgeBaseStore struct{}
func NewKnowledgeBaseStore() *KnowledgeBaseStore { return &KnowledgeBaseStore{} }
// ── KB CRUD ──────────────────────────────────────
func (s *KnowledgeBaseStore) Create(ctx context.Context, kb *models.KnowledgeBase) error {
kb.ID = store.NewID()
now := time.Now().UTC()
kb.CreatedAt = now
kb.UpdatedAt = now
// document_count, chunk_count, total_bytes default to 0 (Go zero values)
_, err := DB.ExecContext(ctx, `
INSERT INTO knowledge_bases (id, name, description, scope, owner_id, team_id, embedding_config, status, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
kb.ID, kb.Name, kb.Description, kb.Scope,
models.NullString(kb.OwnerID), models.NullString(kb.TeamID),
ToJSON(kb.EmbeddingConfig), kb.Status,
now.Format(timeFmt), now.Format(timeFmt),
)
return err
}
func (s *KnowledgeBaseStore) GetByID(ctx context.Context, id string) (*models.KnowledgeBase, error) {
kb, err := scanKB(DB.QueryRowContext(ctx, `
SELECT id, name, description, scope, owner_id, team_id, embedding_config,
document_count, chunk_count, total_bytes, status, discoverable, created_at, updated_at
FROM knowledge_bases WHERE id = ?`, id))
if err != nil {
return nil, err
}
return kb, nil
}
func (s *KnowledgeBaseStore) Update(ctx context.Context, id string, fields map[string]interface{}) error {
b := NewUpdate("knowledge_bases")
for k, v := range fields {
if k == "embedding_config" {
b.SetJSON(k, v)
} else {
b.Set(k, v)
}
}
b.SetExpr("updated_at", nowExpr)
if !b.HasSets() {
return nil
}
b.Where("id", id)
_, err := b.Exec(DB)
return err
}
func (s *KnowledgeBaseStore) Delete(ctx context.Context, id string) error {
_, err := DB.ExecContext(ctx, "DELETE FROM knowledge_bases WHERE id = ?", id)
return err
}
// ── Scoped Listing ──────────────────────────────
func (s *KnowledgeBaseStore) ListForUser(ctx context.Context, userID string, teamIDs []string) ([]models.KnowledgeBase, error) {
q := `
SELECT id, name, description, scope, owner_id, team_id, embedding_config,
document_count, chunk_count, total_bytes, status, discoverable, created_at, updated_at
FROM knowledge_bases
WHERE scope = 'global'
OR (scope = 'personal' AND owner_id = ?)`
args := []interface{}{userID}
if len(teamIDs) > 0 {
placeholders := makeQPlaceholders(len(teamIDs))
q += fmt.Sprintf(` OR (scope = 'team' AND team_id IN (%s))`, placeholders)
for _, tid := range teamIDs {
args = append(args, tid)
}
}
// Group-granted KBs
q += `
OR id IN (
SELECT rg.resource_id FROM resource_grants rg
WHERE rg.resource_type = 'knowledge_base'
AND (
rg.grant_scope = 'global'
OR (rg.grant_scope = 'groups'
AND EXISTS (
SELECT 1 FROM group_members gm
JOIN json_each(rg.granted_groups) je ON je.value = gm.group_id
WHERE gm.user_id = ?
))
)
)`
args = append(args, userID)
q += ` ORDER BY name`
return queryKBs(ctx, q, args...)
}
func (s *KnowledgeBaseStore) ListPersonal(ctx context.Context, userID string) ([]models.KnowledgeBase, error) {
return queryKBs(ctx, `
SELECT id, name, description, scope, owner_id, team_id, embedding_config,
document_count, chunk_count, total_bytes, status, discoverable, created_at, updated_at
FROM knowledge_bases WHERE scope = 'personal' AND owner_id = ? ORDER BY name`, userID)
}
// ── Documents ────────────────────────────────────
func (s *KnowledgeBaseStore) CreateDocument(ctx context.Context, doc *models.KBDocument) error {
doc.ID = store.NewID()
now := time.Now().UTC()
doc.CreatedAt = now
doc.UpdatedAt = now
// chunk_count defaults to 0 (Go zero value)
_, err := DB.ExecContext(ctx, `
INSERT INTO kb_documents (id, kb_id, filename, content_type, size_bytes, storage_key, status, uploaded_by, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
doc.ID, doc.KBID, doc.Filename, doc.ContentType, doc.SizeBytes,
doc.StorageKey, doc.Status, doc.UploadedBy,
now.Format(timeFmt), now.Format(timeFmt),
)
return err
}
func (s *KnowledgeBaseStore) GetDocument(ctx context.Context, id string) (*models.KBDocument, error) {
var doc models.KBDocument
var extractedText, errMsg sql.NullString
err := DB.QueryRowContext(ctx, `
SELECT id, kb_id, filename, content_type, size_bytes, storage_key,
extracted_text, chunk_count, status, error, uploaded_by, created_at, updated_at
FROM kb_documents WHERE id = ?`, id).Scan(
&doc.ID, &doc.KBID, &doc.Filename, &doc.ContentType, &doc.SizeBytes,
&doc.StorageKey, &extractedText, &doc.ChunkCount, &doc.Status,
&errMsg, &doc.UploadedBy, st(&doc.CreatedAt), st(&doc.UpdatedAt),
)
if err != nil {
return nil, err
}
doc.ExtractedText = NullableStringPtr(extractedText)
doc.Error = NullableStringPtr(errMsg)
return &doc, nil
}
func (s *KnowledgeBaseStore) ListDocuments(ctx context.Context, kbID string) ([]models.KBDocument, error) {
rows, err := DB.QueryContext(ctx, `
SELECT id, kb_id, filename, content_type, size_bytes, storage_key,
chunk_count, status, error, uploaded_by, created_at, updated_at
FROM kb_documents WHERE kb_id = ? ORDER BY created_at`, kbID)
if err != nil {
return nil, err
}
defer rows.Close()
var result []models.KBDocument
for rows.Next() {
var doc models.KBDocument
var errMsg sql.NullString
err := rows.Scan(&doc.ID, &doc.KBID, &doc.Filename, &doc.ContentType,
&doc.SizeBytes, &doc.StorageKey, &doc.ChunkCount, &doc.Status,
&errMsg, &doc.UploadedBy, st(&doc.CreatedAt), st(&doc.UpdatedAt))
if err != nil {
return nil, err
}
doc.Error = NullableStringPtr(errMsg)
result = append(result, doc)
}
return result, rows.Err()
}
func (s *KnowledgeBaseStore) UpdateDocumentStatus(ctx context.Context, id string, status string, errMsg *string) error {
_, err := DB.ExecContext(ctx, `
UPDATE kb_documents SET status = ?, error = ?, updated_at = datetime('now')
WHERE id = ?`, status, models.NullString(errMsg), id)
return err
}
func (s *KnowledgeBaseStore) UpdateDocumentText(ctx context.Context, id string, text string, chunkCount int) error {
_, err := DB.ExecContext(ctx, `
UPDATE kb_documents SET extracted_text = ?, chunk_count = ?, updated_at = datetime('now')
WHERE id = ?`, text, chunkCount, id)
return err
}
func (s *KnowledgeBaseStore) UpdateDocumentStorageKey(ctx context.Context, id string, storageKey string) error {
_, err := DB.ExecContext(ctx, `
UPDATE kb_documents SET storage_key = ? WHERE id = ?`, storageKey, id)
return err
}
func (s *KnowledgeBaseStore) DeleteDocument(ctx context.Context, id string) (*models.KBDocument, error) {
var doc models.KBDocument
var errMsg sql.NullString
err := DB.QueryRowContext(ctx, `
DELETE FROM kb_documents WHERE id = ?
RETURNING id, kb_id, filename, storage_key, status, error`,
id).Scan(&doc.ID, &doc.KBID, &doc.Filename, &doc.StorageKey, &doc.Status, &errMsg)
if err != nil {
return nil, err
}
doc.Error = NullableStringPtr(errMsg)
return &doc, nil
}
// ── Chunks ───────────────────────────────────────
func (s *KnowledgeBaseStore) InsertChunks(ctx context.Context, chunks []models.KBChunk) error {
if len(chunks) == 0 {
return nil
}
// Batch insert using multi-row INSERT.
// Embeddings stored as JSON text (e.g. "[0.1,0.2,...]") for app-level cosine similarity.
valueParts := make([]string, 0, len(chunks))
args := make([]interface{}, 0, len(chunks)*8)
for _, c := range chunks {
c.ID = store.NewID()
valueParts = append(valueParts, "(?, ?, ?, ?, ?, ?, ?, ?)")
var embJSON interface{}
if len(c.Embedding) > 0 {
embJSON = ToJSON(c.Embedding)
}
args = append(args, c.ID, c.KBID, c.DocumentID, c.ChunkIndex,
c.Content, c.TokenCount, embJSON, ToJSON(c.Metadata))
}
q := fmt.Sprintf(`INSERT INTO kb_chunks (id, kb_id, document_id, chunk_index, content, token_count, embedding, metadata)
VALUES %s`, strings.Join(valueParts, ", "))
_, err := DB.ExecContext(ctx, q, args...)
return err
}
func (s *KnowledgeBaseStore) DeleteChunksForDocument(ctx context.Context, documentID string) error {
_, err := DB.ExecContext(ctx, "DELETE FROM kb_chunks WHERE document_id = ?", documentID)
return err
}
// SimilaritySearch performs cosine similarity search in Go.
// Loads candidate chunks from the specified KBs, decodes their JSON-stored
// embeddings, computes cosine similarity against queryVec, and returns the
// top results above threshold. For SQLite-scale deployments (single-user,
// edge, dev) this is perfectly adequate without requiring CGO/sqlite-vec.
func (s *KnowledgeBaseStore) SimilaritySearch(ctx context.Context, kbIDs []string, queryVec []float64, threshold float64, limit int) ([]models.KBSearchResult, error) {
if len(kbIDs) == 0 || len(queryVec) == 0 {
return nil, nil
}
if limit <= 0 {
limit = 5
}
// Load candidate chunks with embeddings
placeholders := makeQPlaceholders(len(kbIDs))
q := fmt.Sprintf(`
SELECT c.content, c.embedding, c.metadata, d.filename, kb.name
FROM kb_chunks c
JOIN kb_documents d ON c.document_id = d.id
JOIN knowledge_bases kb ON c.kb_id = kb.id
WHERE c.kb_id IN (%s)
AND c.embedding IS NOT NULL`, placeholders)
args := make([]interface{}, len(kbIDs))
for i, id := range kbIDs {
args[i] = id
}
rows, err := DB.QueryContext(ctx, q, args...)
if err != nil {
return nil, err
}
defer rows.Close()
type candidate struct {
result models.KBSearchResult
similarity float64
}
var candidates []candidate
for rows.Next() {
var content, embJSON, filename, kbName string
var metaJSON sql.NullString
if err := rows.Scan(&content, &embJSON, &metaJSON, &filename, &kbName); err != nil {
return nil, err
}
// Decode embedding from JSON
var embedding []float64
if err := json.Unmarshal([]byte(embJSON), &embedding); err != nil {
continue // skip chunks with malformed embeddings
}
sim := cosineSimilarity(queryVec, embedding)
if sim > threshold {
r := models.KBSearchResult{
Content: content,
Filename: filename,
KBName: kbName,
Similarity: sim,
}
if metaJSON.Valid {
ScanJSON(metaJSON.String, &r.Metadata)
}
candidates = append(candidates, candidate{result: r, similarity: sim})
}
}
if err := rows.Err(); err != nil {
return nil, err
}
// Sort by similarity descending
sort.Slice(candidates, func(i, j int) bool {
return candidates[i].similarity > candidates[j].similarity
})
// Cap at limit
if len(candidates) > limit {
candidates = candidates[:limit]
}
results := make([]models.KBSearchResult, len(candidates))
for i, c := range candidates {
results[i] = c.result
}
return results, nil
}
// cosineSimilarity computes cosine similarity between two vectors.
// Returns 0 if either vector is zero-length or all-zeros.
func cosineSimilarity(a, b []float64) float64 {
if len(a) != len(b) || len(a) == 0 {
return 0
}
var dot, normA, normB float64
for i := range a {
dot += a[i] * b[i]
normA += a[i] * a[i]
normB += b[i] * b[i]
}
if normA == 0 || normB == 0 {
return 0
}
return dot / (math.Sqrt(normA) * math.Sqrt(normB))
}
// ── Channel Links ─────────────────────────────────
func (s *KnowledgeBaseStore) SetChannelKBs(ctx context.Context, channelID string, kbIDs []string) error {
tx, err := DB.BeginTx(ctx, nil)
if err != nil {
return err
}
defer tx.Rollback()
_, err = tx.ExecContext(ctx, "DELETE FROM channel_knowledge_bases WHERE channel_id = ?", channelID)
if err != nil {
return err
}
for _, kbID := range kbIDs {
_, err = tx.ExecContext(ctx, `
INSERT INTO channel_knowledge_bases (channel_id, kb_id, enabled) VALUES (?, ?, 1)`,
channelID, kbID)
if err != nil {
return err
}
}
return tx.Commit()
}
func (s *KnowledgeBaseStore) GetChannelKBs(ctx context.Context, channelID string) ([]models.ChannelKB, error) {
rows, err := DB.QueryContext(ctx, `
SELECT ckb.kb_id, kb.name, ckb.enabled, kb.document_count
FROM channel_knowledge_bases ckb
JOIN knowledge_bases kb ON ckb.kb_id = kb.id
WHERE ckb.channel_id = ?
ORDER BY kb.name`, channelID)
if err != nil {
return nil, err
}
defer rows.Close()
var result []models.ChannelKB
for rows.Next() {
var ckb models.ChannelKB
if err := rows.Scan(&ckb.KBID, &ckb.KBName, &ckb.Enabled, &ckb.DocumentCount); err != nil {
return nil, err
}
result = append(result, ckb)
}
return result, rows.Err()
}
func (s *KnowledgeBaseStore) GetActiveKBIDs(ctx context.Context, channelID string, userID string, teamIDs []string) ([]string, error) {
q := `
SELECT ckb.kb_id
FROM channel_knowledge_bases ckb
JOIN knowledge_bases kb ON ckb.kb_id = kb.id
WHERE ckb.channel_id = ? AND ckb.enabled = 1
AND (
kb.scope = 'global'
OR (kb.scope = 'personal' AND kb.owner_id = ?)`
args := []interface{}{channelID, userID}
if len(teamIDs) > 0 {
placeholders := makeQPlaceholders(len(teamIDs))
q += fmt.Sprintf(` OR (kb.scope = 'team' AND kb.team_id IN (%s))`, placeholders)
for _, tid := range teamIDs {
args = append(args, tid)
}
}
q += `)`
rows, err := DB.QueryContext(ctx, q, args...)
if err != nil {
return nil, err
}
defer rows.Close()
var ids []string
for rows.Next() {
var id string
if err := rows.Scan(&id); err != nil {
return nil, err
}
ids = append(ids, id)
}
return ids, rows.Err()
}
// ── Stats ────────────────────────────────────────
func (s *KnowledgeBaseStore) UpdateStats(ctx context.Context, kbID string) error {
_, err := DB.ExecContext(ctx, `
UPDATE knowledge_bases SET
document_count = (SELECT COUNT(*) FROM kb_documents WHERE kb_id = ? AND status != 'error'),
chunk_count = (SELECT COUNT(*) FROM kb_chunks WHERE kb_id = ?),
total_bytes = COALESCE((SELECT SUM(size_bytes) FROM kb_documents WHERE kb_id = ?), 0),
updated_at = datetime('now')
WHERE id = ?`, kbID, kbID, kbID, kbID)
return err
}
// ── Discoverable Management (v0.17.0) ────────────
func (s *KnowledgeBaseStore) SetDiscoverable(ctx context.Context, kbID string, discoverable bool) error {
_, err := DB.ExecContext(ctx, `
UPDATE knowledge_bases SET discoverable = ?, updated_at = datetime('now')
WHERE id = ?`, discoverable, kbID)
return err
}
// GetActiveKBIDsWithPersona returns KB IDs accessible for a channel, including
// KBs bound to the active persona.
func (s *KnowledgeBaseStore) GetActiveKBIDsWithPersona(ctx context.Context, channelID string, userID string, teamIDs []string, personaID string) ([]string, error) {
q := `
SELECT ckb.kb_id
FROM channel_knowledge_bases ckb
JOIN knowledge_bases kb ON ckb.kb_id = kb.id
WHERE ckb.channel_id = ? AND ckb.enabled = 1
AND (
kb.scope = 'global'
OR (kb.scope = 'personal' AND kb.owner_id = ?)`
args := []interface{}{channelID, userID}
if len(teamIDs) > 0 {
placeholders := makeQPlaceholders(len(teamIDs))
q += fmt.Sprintf(` OR (kb.scope = 'team' AND kb.team_id IN (%s))`, placeholders)
for _, tid := range teamIDs {
args = append(args, tid)
}
}
q += `)`
// UNION with persona-bound KBs (bypass discoverable check)
if personaID != "" {
q += `
UNION
SELECT pkb.kb_id
FROM persona_knowledge_bases pkb
JOIN knowledge_bases kb ON pkb.kb_id = kb.id
WHERE pkb.persona_id = ?
AND kb.chunk_count > 0`
args = append(args, personaID)
}
rows, err := DB.QueryContext(ctx, q, args...)
if err != nil {
return nil, err
}
defer rows.Close()
seen := make(map[string]bool)
var ids []string
for rows.Next() {
var id string
if err := rows.Scan(&id); err != nil {
return nil, err
}
if !seen[id] {
seen[id] = true
ids = append(ids, id)
}
}
return ids, rows.Err()
}
// ListDiscoverable returns KBs the user can see AND that are discoverable.
func (s *KnowledgeBaseStore) ListDiscoverable(ctx context.Context, userID string, teamIDs []string) ([]models.KnowledgeBase, error) {
q := `
SELECT id, name, description, scope, owner_id, team_id, embedding_config,
document_count, chunk_count, total_bytes, status, discoverable, created_at, updated_at
FROM knowledge_bases
WHERE discoverable = 1
AND (
scope = 'global'
OR (scope = 'personal' AND owner_id = ?)`
args := []interface{}{userID}
if len(teamIDs) > 0 {
placeholders := makeQPlaceholders(len(teamIDs))
q += fmt.Sprintf(` OR (scope = 'team' AND team_id IN (%s))`, placeholders)
for _, tid := range teamIDs {
args = append(args, tid)
}
}
// Group-granted KBs
q += `
OR id IN (
SELECT rg.resource_id FROM resource_grants rg
WHERE rg.resource_type = 'knowledge_base'
AND (
rg.grant_scope = 'global'
OR (rg.grant_scope = 'groups'
AND EXISTS (
SELECT 1 FROM group_members gm
JOIN json_each(rg.granted_groups) je ON je.value = gm.group_id
WHERE gm.user_id = ?
))
)
)`
args = append(args, userID)
q += `) ORDER BY name`
return queryKBs(ctx, q, args...)
}
// ── Scan Helpers ─────────────────────────────────
func scanKB(row *sql.Row) (*models.KnowledgeBase, error) {
var kb models.KnowledgeBase
var ownerID, teamID sql.NullString
var embCfgJSON string
err := row.Scan(
&kb.ID, &kb.Name, &kb.Description, &kb.Scope,
&ownerID, &teamID, &embCfgJSON,
&kb.DocumentCount, &kb.ChunkCount, &kb.TotalBytes,
&kb.Status, &kb.Discoverable, st(&kb.CreatedAt), st(&kb.UpdatedAt),
)
if err != nil {
return nil, err
}
kb.OwnerID = NullableStringPtr(ownerID)
kb.TeamID = NullableStringPtr(teamID)
ScanJSON(embCfgJSON, &kb.EmbeddingConfig)
return &kb, nil
}
func queryKBs(ctx context.Context, query string, args ...interface{}) ([]models.KnowledgeBase, error) {
rows, err := DB.QueryContext(ctx, query, args...)
if err != nil {
return nil, err
}
defer rows.Close()
var result []models.KnowledgeBase
for rows.Next() {
var kb models.KnowledgeBase
var ownerID, teamID sql.NullString
var embCfgJSON string
err := rows.Scan(
&kb.ID, &kb.Name, &kb.Description, &kb.Scope,
&ownerID, &teamID, &embCfgJSON,
&kb.DocumentCount, &kb.ChunkCount, &kb.TotalBytes,
&kb.Status, &kb.Discoverable, st(&kb.CreatedAt), st(&kb.UpdatedAt),
)
if err != nil {
return nil, err
}
kb.OwnerID = NullableStringPtr(ownerID)
kb.TeamID = NullableStringPtr(teamID)
ScanJSON(embCfgJSON, &kb.EmbeddingConfig)
result = append(result, kb)
}
return result, rows.Err()
}
// makeQPlaceholders returns "?,?,?" for n items.
func makeQPlaceholders(n int) string {
if n <= 0 {
return ""
}
return strings.TrimRight(strings.Repeat("?,", n), ",")
}