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), ",") }