search.go 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499
  1. package cache
  2. import (
  3. "context"
  4. "fmt"
  5. "os"
  6. "path/filepath"
  7. "strings"
  8. "sync"
  9. "time"
  10. "github.com/blevesearch/bleve/v2"
  11. "github.com/blevesearch/bleve/v2/analysis/lang/en"
  12. "github.com/blevesearch/bleve/v2/mapping"
  13. "github.com/blevesearch/bleve/v2/search/query"
  14. "github.com/uozi-tech/cosy/logger"
  15. )
  16. // SearchDocument represents a document in the search index
  17. type SearchDocument struct {
  18. ID string `json:"id"`
  19. Type string `json:"type"` // "site", "stream", or "config"
  20. Name string `json:"name"` // extracted from filename
  21. Path string `json:"path"` // file path
  22. Content string `json:"content"` // file content
  23. UpdatedAt time.Time `json:"updated_at"`
  24. }
  25. // SearchResult represents a search result
  26. type SearchResult struct {
  27. Document SearchDocument `json:"document"`
  28. Score float64 `json:"score"`
  29. }
  30. // SearchIndexer manages the Bleve search index
  31. type SearchIndexer struct {
  32. index bleve.Index
  33. indexPath string
  34. indexMutex sync.RWMutex
  35. ctx context.Context
  36. cancel context.CancelFunc
  37. cleanupOnce sync.Once
  38. }
  39. var (
  40. searchIndexer *SearchIndexer
  41. searchIndexerOnce sync.Once
  42. )
  43. // GetSearchIndexer returns the singleton search indexer instance
  44. func GetSearchIndexer() *SearchIndexer {
  45. searchIndexerOnce.Do(func() {
  46. // Create a temporary directory for the index
  47. tempDir, err := os.MkdirTemp("", "nginx-ui-search-index-*")
  48. if err != nil {
  49. logger.Fatalf("Failed to create temp directory for search index: %v", err)
  50. }
  51. searchIndexer = &SearchIndexer{
  52. indexPath: tempDir,
  53. }
  54. })
  55. return searchIndexer
  56. }
  57. // InitSearchIndex initializes the search index
  58. func InitSearchIndex(ctx context.Context) error {
  59. indexer := GetSearchIndexer()
  60. return indexer.Initialize(ctx)
  61. }
  62. // Initialize sets up the Bleve search index
  63. func (si *SearchIndexer) Initialize(ctx context.Context) error {
  64. si.indexMutex.Lock()
  65. defer si.indexMutex.Unlock()
  66. // Create a derived context for cleanup
  67. si.ctx, si.cancel = context.WithCancel(ctx)
  68. // Check if context is cancelled
  69. select {
  70. case <-ctx.Done():
  71. return ctx.Err()
  72. default:
  73. }
  74. // Try to open existing index, create new if it fails
  75. var err error
  76. si.index, err = bleve.Open(si.indexPath)
  77. if err != nil {
  78. // Check context again before creating new index
  79. select {
  80. case <-ctx.Done():
  81. return ctx.Err()
  82. default:
  83. }
  84. logger.Info("Creating new search index at:", si.indexPath)
  85. si.index, err = bleve.New(si.indexPath, si.createIndexMapping())
  86. if err != nil {
  87. return fmt.Errorf("failed to create search index: %w", err)
  88. }
  89. }
  90. // Register callback for config scanning
  91. RegisterCallback(si.handleConfigScan)
  92. // Start cleanup goroutine
  93. go si.watchContext()
  94. logger.Info("Search index initialized successfully")
  95. return nil
  96. }
  97. // watchContext monitors the context and cleans up when it's cancelled
  98. func (si *SearchIndexer) watchContext() {
  99. <-si.ctx.Done()
  100. si.cleanup()
  101. }
  102. // cleanup closes the index and removes the temporary directory
  103. func (si *SearchIndexer) cleanup() {
  104. si.cleanupOnce.Do(func() {
  105. logger.Info("Cleaning up search index...")
  106. si.indexMutex.Lock()
  107. defer si.indexMutex.Unlock()
  108. if si.index != nil {
  109. si.index.Close()
  110. si.index = nil
  111. }
  112. // Remove the temporary directory
  113. if err := os.RemoveAll(si.indexPath); err != nil {
  114. logger.Error("Failed to remove search index directory:", err)
  115. } else {
  116. logger.Info("Search index directory removed successfully")
  117. }
  118. })
  119. }
  120. // createIndexMapping creates the mapping for the search index
  121. func (si *SearchIndexer) createIndexMapping() mapping.IndexMapping {
  122. docMapping := bleve.NewDocumentMapping()
  123. // Text fields with standard analyzer
  124. textField := bleve.NewTextFieldMapping()
  125. textField.Analyzer = en.AnalyzerName
  126. textField.Store = true
  127. textField.Index = true
  128. // Keyword fields for exact match
  129. keywordField := bleve.NewKeywordFieldMapping()
  130. keywordField.Store = true
  131. keywordField.Index = true
  132. // Date field
  133. dateField := bleve.NewDateTimeFieldMapping()
  134. dateField.Store = true
  135. dateField.Index = true
  136. // Map fields to types
  137. fieldMappings := map[string]*mapping.FieldMapping{
  138. "id": keywordField,
  139. "type": keywordField,
  140. "path": keywordField,
  141. "name": textField,
  142. "content": textField,
  143. "updated_at": dateField,
  144. }
  145. for field, fieldMapping := range fieldMappings {
  146. docMapping.AddFieldMappingsAt(field, fieldMapping)
  147. }
  148. indexMapping := bleve.NewIndexMapping()
  149. indexMapping.DefaultMapping = docMapping
  150. indexMapping.DefaultAnalyzer = en.AnalyzerName
  151. return indexMapping
  152. }
  153. // handleConfigScan processes scanned config files and indexes them
  154. func (si *SearchIndexer) handleConfigScan(configPath string, content []byte) error {
  155. docType := si.determineConfigType(configPath)
  156. if docType == "" {
  157. return nil // Skip unsupported file types
  158. }
  159. doc := SearchDocument{
  160. ID: configPath,
  161. Type: docType,
  162. Name: filepath.Base(configPath),
  163. Path: configPath,
  164. Content: string(content),
  165. UpdatedAt: time.Now(),
  166. }
  167. return si.IndexDocument(doc)
  168. }
  169. // determineConfigType determines the type of config file based on path
  170. func (si *SearchIndexer) determineConfigType(configPath string) string {
  171. normalizedPath := filepath.ToSlash(configPath)
  172. switch {
  173. case strings.Contains(normalizedPath, "sites-available") || strings.Contains(normalizedPath, "sites-enabled"):
  174. return "site"
  175. case strings.Contains(normalizedPath, "streams-available") || strings.Contains(normalizedPath, "streams-enabled"):
  176. return "stream"
  177. default:
  178. return "config"
  179. }
  180. }
  181. // IndexDocument indexes a single document
  182. func (si *SearchIndexer) IndexDocument(doc SearchDocument) error {
  183. si.indexMutex.RLock()
  184. defer si.indexMutex.RUnlock()
  185. if si.index == nil {
  186. return fmt.Errorf("search index not initialized")
  187. }
  188. // logger.Debugf("Indexing document: ID=%s, Type=%s, Name=%s, Path=%s",
  189. // doc.ID, doc.Type, doc.Name, doc.Path)
  190. return si.index.Index(doc.ID, doc)
  191. }
  192. // Search performs a search query
  193. func (si *SearchIndexer) Search(ctx context.Context, queryStr string, limit int) ([]SearchResult, error) {
  194. return si.searchWithType(ctx, queryStr, "", limit)
  195. }
  196. // SearchByType performs a search filtered by document type
  197. func (si *SearchIndexer) SearchByType(ctx context.Context, queryStr string, docType string, limit int) ([]SearchResult, error) {
  198. return si.searchWithType(ctx, queryStr, docType, limit)
  199. }
  200. // searchWithType performs the actual search with optional type filtering
  201. func (si *SearchIndexer) searchWithType(ctx context.Context, queryStr string, docType string, limit int) ([]SearchResult, error) {
  202. si.indexMutex.RLock()
  203. defer si.indexMutex.RUnlock()
  204. // Check if context is cancelled
  205. select {
  206. case <-ctx.Done():
  207. return nil, ctx.Err()
  208. default:
  209. }
  210. if si.index == nil {
  211. return nil, fmt.Errorf("search index not initialized")
  212. }
  213. if limit <= 0 {
  214. limit = 500 // Increase default limit to handle more results
  215. }
  216. query := si.buildQuery(queryStr, docType)
  217. searchRequest := bleve.NewSearchRequest(query)
  218. searchRequest.Size = limit
  219. searchRequest.Fields = []string{"*"}
  220. // Use a channel to handle search with context cancellation
  221. type searchResult struct {
  222. result *bleve.SearchResult
  223. err error
  224. }
  225. resultChan := make(chan searchResult, 1)
  226. go func() {
  227. result, err := si.index.Search(searchRequest)
  228. resultChan <- searchResult{result: result, err: err}
  229. }()
  230. // Wait for search result or context cancellation
  231. select {
  232. case <-ctx.Done():
  233. return nil, ctx.Err()
  234. case res := <-resultChan:
  235. if res.err != nil {
  236. return nil, fmt.Errorf("search execution failed: %w", res.err)
  237. }
  238. results := si.convertResults(res.result)
  239. // Debug log the search execution
  240. logger.Debugf("Search index query '%s' (type: %s, limit: %d) returned %d results",
  241. queryStr, docType, limit, len(results))
  242. return results, nil
  243. }
  244. }
  245. // buildQuery builds a search query with optional type filtering
  246. func (si *SearchIndexer) buildQuery(queryStr string, docType string) query.Query {
  247. mainQuery := bleve.NewBooleanQuery()
  248. // Add type filter if specified
  249. if docType != "" {
  250. typeQuery := bleve.NewTermQuery(docType)
  251. typeQuery.SetField("type")
  252. mainQuery.AddMust(typeQuery)
  253. }
  254. // Add text search across name and content fields only
  255. textQuery := bleve.NewBooleanQuery()
  256. searchFields := []string{"name", "content"}
  257. for _, field := range searchFields {
  258. // Create a boolean query for this field to combine multiple query types
  259. fieldQuery := bleve.NewBooleanQuery()
  260. // 1. Exact match query (highest priority)
  261. matchQuery := bleve.NewMatchQuery(queryStr)
  262. matchQuery.SetField(field)
  263. matchQuery.SetBoost(3.0) // Higher boost for exact matches
  264. fieldQuery.AddShould(matchQuery)
  265. // 2. Prefix query for partial matches (e.g., "access" matches "access_log")
  266. prefixQuery := bleve.NewPrefixQuery(queryStr)
  267. prefixQuery.SetField(field)
  268. prefixQuery.SetBoost(2.0) // Medium boost for prefix matches
  269. fieldQuery.AddShould(prefixQuery)
  270. // 3. Wildcard query for more flexible matching
  271. wildcardQuery := bleve.NewWildcardQuery("*" + queryStr + "*")
  272. wildcardQuery.SetField(field)
  273. wildcardQuery.SetBoost(1.5) // Lower boost for wildcard matches
  274. fieldQuery.AddShould(wildcardQuery)
  275. // 4. Fuzzy match query (allows 1 character difference)
  276. fuzzyQuery := bleve.NewFuzzyQuery(queryStr)
  277. fuzzyQuery.SetField(field)
  278. fuzzyQuery.SetFuzziness(1)
  279. fuzzyQuery.SetBoost(1.0) // Lowest boost for fuzzy matches
  280. fieldQuery.AddShould(fuzzyQuery)
  281. textQuery.AddShould(fieldQuery)
  282. }
  283. if docType != "" {
  284. mainQuery.AddMust(textQuery)
  285. } else {
  286. return textQuery
  287. }
  288. return mainQuery
  289. }
  290. // convertResults converts Bleve search results to our SearchResult format
  291. func (si *SearchIndexer) convertResults(searchResult *bleve.SearchResult) []SearchResult {
  292. results := make([]SearchResult, 0, len(searchResult.Hits))
  293. for _, hit := range searchResult.Hits {
  294. doc := SearchDocument{
  295. ID: si.getStringField(hit.Fields, "id"),
  296. Type: si.getStringField(hit.Fields, "type"),
  297. Name: si.getStringField(hit.Fields, "name"),
  298. Path: si.getStringField(hit.Fields, "path"),
  299. Content: si.getStringField(hit.Fields, "content"),
  300. }
  301. // Parse updated_at if present
  302. if updatedAtStr := si.getStringField(hit.Fields, "updated_at"); updatedAtStr != "" {
  303. if updatedAt, err := time.Parse(time.RFC3339, updatedAtStr); err == nil {
  304. doc.UpdatedAt = updatedAt
  305. }
  306. }
  307. results = append(results, SearchResult{
  308. Document: doc,
  309. Score: hit.Score,
  310. })
  311. }
  312. return results
  313. }
  314. // getStringField safely gets a string field from search results
  315. func (si *SearchIndexer) getStringField(fields map[string]interface{}, fieldName string) string {
  316. if value, ok := fields[fieldName]; ok {
  317. if str, ok := value.(string); ok {
  318. return str
  319. }
  320. }
  321. return ""
  322. }
  323. // DeleteDocument removes a document from the index
  324. func (si *SearchIndexer) DeleteDocument(docID string) error {
  325. si.indexMutex.RLock()
  326. defer si.indexMutex.RUnlock()
  327. if si.index == nil {
  328. return fmt.Errorf("search index not initialized")
  329. }
  330. return si.index.Delete(docID)
  331. }
  332. // RebuildIndex rebuilds the entire search index
  333. func (si *SearchIndexer) RebuildIndex(ctx context.Context) error {
  334. si.indexMutex.Lock()
  335. defer si.indexMutex.Unlock()
  336. // Check if context is cancelled
  337. select {
  338. case <-ctx.Done():
  339. return ctx.Err()
  340. default:
  341. }
  342. if si.index != nil {
  343. si.index.Close()
  344. }
  345. // Check context before removing old index
  346. select {
  347. case <-ctx.Done():
  348. return ctx.Err()
  349. default:
  350. }
  351. // Remove old index
  352. if err := os.RemoveAll(si.indexPath); err != nil {
  353. logger.Error("Failed to remove old index:", err)
  354. }
  355. // Check context before creating new index
  356. select {
  357. case <-ctx.Done():
  358. return ctx.Err()
  359. default:
  360. }
  361. // Create new index
  362. var err error
  363. si.index, err = bleve.New(si.indexPath, si.createIndexMapping())
  364. if err != nil {
  365. return fmt.Errorf("failed to create new index: %w", err)
  366. }
  367. logger.Info("Search index rebuilt successfully")
  368. return nil
  369. }
  370. // GetIndexStats returns statistics about the search index
  371. func (si *SearchIndexer) GetIndexStats() (map[string]interface{}, error) {
  372. si.indexMutex.RLock()
  373. defer si.indexMutex.RUnlock()
  374. if si.index == nil {
  375. return nil, fmt.Errorf("search index not initialized")
  376. }
  377. docCount, err := si.index.DocCount()
  378. if err != nil {
  379. return nil, err
  380. }
  381. return map[string]interface{}{
  382. "document_count": docCount,
  383. "index_path": si.indexPath,
  384. }, nil
  385. }
  386. // Close closes the search index and triggers cleanup
  387. func (si *SearchIndexer) Close() error {
  388. if si.cancel != nil {
  389. si.cancel()
  390. }
  391. si.cleanup()
  392. return nil
  393. }
  394. // Convenience functions for different search types
  395. // SearchSites searches only site configurations
  396. func SearchSites(ctx context.Context, query string, limit int) ([]SearchResult, error) {
  397. return GetSearchIndexer().SearchByType(ctx, query, "site", limit)
  398. }
  399. // SearchStreams searches only stream configurations
  400. func SearchStreams(ctx context.Context, query string, limit int) ([]SearchResult, error) {
  401. return GetSearchIndexer().SearchByType(ctx, query, "stream", limit)
  402. }
  403. // SearchConfigs searches only general configurations
  404. func SearchConfigs(ctx context.Context, query string, limit int) ([]SearchResult, error) {
  405. return GetSearchIndexer().SearchByType(ctx, query, "config", limit)
  406. }
  407. // SearchAll searches across all configuration types
  408. func SearchAll(ctx context.Context, query string, limit int) ([]SearchResult, error) {
  409. return GetSearchIndexer().Search(ctx, query, limit)
  410. }