appview/indexer: add pulls indexer #708

+3

appview/indexer/indexer.go

··· 6 6 7 7 "tangled.org/core/appview/db" 8 8 issues_indexer "tangled.org/core/appview/indexer/issues" 9 9 + pulls_indexer "tangled.org/core/appview/indexer/pulls" 9 10 "tangled.org/core/appview/notify" 10 11 tlog "tangled.org/core/log" 11 12 ) 12 13 13 14 type Indexer struct { 14 15 Issues *issues_indexer.Indexer 16 16 + Pulls *pulls_indexer.Indexer 15 17 logger *slog.Logger 16 18 notify.BaseNotifier 17 19 } ··· 29 31 func (ix *Indexer) Init(ctx context.Context, db *db.DB) error { 30 32 ctx = tlog.IntoContext(ctx, ix.logger) 31 33 ix.Issues.Init(ctx, db) 34 34 + ix.Pulls.Init(ctx, db) 32 35 return nil 33 36 }

+27

appview/indexer/notifier.go

··· 36 36 l.Error("failed to delete an issue", "err", err) 37 37 } 38 38 } 39 39 + 40 40 + func (ix *Indexer) NewPull(ctx context.Context, pull *models.Pull) { 41 41 + l := log.FromContext(ctx).With("notifier", "indexer", "pull", pull) 42 42 + l.Debug("indexing new pr") 43 43 + err := ix.Pulls.Index(ctx, pull) 44 44 + if err != nil { 45 45 + l.Error("failed to index a pr", "err", err) 46 46 + } 47 47 + } 48 48 + 49 49 + func (ix *Indexer) NewPullMerged(ctx context.Context, pull *models.Pull) { 50 50 + l := log.FromContext(ctx).With("notifier", "indexer", "pull", pull) 51 51 + l.Debug("updating a pr") 52 52 + err := ix.Pulls.Index(ctx, pull) 53 53 + if err != nil { 54 54 + l.Error("failed to index a pr", "err", err) 55 55 + } 56 56 + } 57 57 + 58 58 + func (ix *Indexer) NewPullClosed(ctx context.Context, pull *models.Pull) { 59 59 + l := log.FromContext(ctx).With("notifier", "indexer", "pull", pull) 60 60 + l.Debug("updating a pr") 61 61 + err := ix.Pulls.Index(ctx, pull) 62 62 + if err != nil { 63 63 + l.Error("failed to index a pr", "err", err) 64 64 + } 65 65 + }

+255

appview/indexer/pulls/indexer.go

··· 1 1 + // heavily inspired by gitea's model (basically copy-pasted) 2 2 + package pulls_indexer 3 3 + 4 4 + import ( 5 5 + "context" 6 6 + "errors" 7 7 + "log" 8 8 + "os" 9 9 + 10 10 + "github.com/blevesearch/bleve/v2" 11 11 + "github.com/blevesearch/bleve/v2/analysis/analyzer/custom" 12 12 + "github.com/blevesearch/bleve/v2/analysis/token/camelcase" 13 13 + "github.com/blevesearch/bleve/v2/analysis/token/lowercase" 14 14 + "github.com/blevesearch/bleve/v2/analysis/token/unicodenorm" 15 15 + "github.com/blevesearch/bleve/v2/analysis/tokenizer/unicode" 16 16 + "github.com/blevesearch/bleve/v2/index/upsidedown" 17 17 + "github.com/blevesearch/bleve/v2/mapping" 18 18 + "github.com/blevesearch/bleve/v2/search/query" 19 19 + "tangled.org/core/appview/db" 20 20 + "tangled.org/core/appview/indexer/base36" 21 21 + "tangled.org/core/appview/indexer/bleve" 22 22 + "tangled.org/core/appview/models" 23 23 + tlog "tangled.org/core/log" 24 24 + ) 25 25 + 26 26 + const ( 27 27 + pullIndexerAnalyzer = "pullIndexer" 28 28 + pullIndexerDocType = "pullIndexerDocType" 29 29 + 30 30 + unicodeNormalizeName = "uicodeNormalize" 31 31 + ) 32 32 + 33 33 + type Indexer struct { 34 34 + indexer bleve.Index 35 35 + path string 36 36 + } 37 37 + 38 38 + func NewIndexer(indexDir string) *Indexer { 39 39 + return &Indexer{ 40 40 + path: indexDir, 41 41 + } 42 42 + } 43 43 + 44 44 + // Init initializes the indexer 45 45 + func (ix *Indexer) Init(ctx context.Context, e db.Execer) { 46 46 + l := tlog.FromContext(ctx) 47 47 + existed, err := ix.intialize(ctx) 48 48 + if err != nil { 49 49 + log.Fatalln("failed to initialize pull indexer", err) 50 50 + } 51 51 + if !existed { 52 52 + l.Debug("Populating the pull indexer") 53 53 + err := PopulateIndexer(ctx, ix, e) 54 54 + if err != nil { 55 55 + log.Fatalln("failed to populate pull indexer", err) 56 56 + } 57 57 + } 58 58 + l.Info("Initialized the pull indexer") 59 59 + } 60 60 + 61 61 + func generatePullIndexMapping() (mapping.IndexMapping, error) { 62 62 + mapping := bleve.NewIndexMapping() 63 63 + docMapping := bleve.NewDocumentMapping() 64 64 + 65 65 + textFieldMapping := bleve.NewTextFieldMapping() 66 66 + textFieldMapping.Store = false 67 67 + textFieldMapping.IncludeInAll = false 68 68 + 69 69 + keywordFieldMapping := bleve.NewKeywordFieldMapping() 70 70 + keywordFieldMapping.Store = false 71 71 + keywordFieldMapping.IncludeInAll = false 72 72 + 73 73 + // numericFieldMapping := bleve.NewNumericFieldMapping() 74 74 + 75 75 + docMapping.AddFieldMappingsAt("title", textFieldMapping) 76 76 + docMapping.AddFieldMappingsAt("body", textFieldMapping) 77 77 + 78 78 + docMapping.AddFieldMappingsAt("repo_at", keywordFieldMapping) 79 79 + docMapping.AddFieldMappingsAt("state", keywordFieldMapping) 80 80 + 81 81 + err := mapping.AddCustomTokenFilter(unicodeNormalizeName, map[string]any{ 82 82 + "type": unicodenorm.Name, 83 83 + "form": unicodenorm.NFC, 84 84 + }) 85 85 + if err != nil { 86 86 + return nil, err 87 87 + } 88 88 + 89 89 + err = mapping.AddCustomAnalyzer(pullIndexerAnalyzer, map[string]any{ 90 90 + "type": custom.Name, 91 91 + "char_filters": []string{}, 92 92 + "tokenizer": unicode.Name, 93 93 + "token_filters": []string{unicodeNormalizeName, camelcase.Name, lowercase.Name}, 94 94 + }) 95 95 + if err != nil { 96 96 + return nil, err 97 97 + } 98 98 + 99 99 + mapping.DefaultAnalyzer = pullIndexerAnalyzer 100 100 + mapping.AddDocumentMapping(pullIndexerDocType, docMapping) 101 101 + mapping.AddDocumentMapping("_all", bleve.NewDocumentDisabledMapping()) 102 102 + mapping.DefaultMapping = bleve.NewDocumentDisabledMapping() 103 103 + 104 104 + return mapping, nil 105 105 + } 106 106 + 107 107 + func (ix *Indexer) intialize(ctx context.Context) (bool, error) { 108 108 + if ix.indexer != nil { 109 109 + return false, errors.New("indexer is already initialized") 110 110 + } 111 111 + 112 112 + indexer, err := openIndexer(ctx, ix.path) 113 113 + if err != nil { 114 114 + return false, err 115 115 + } 116 116 + if indexer != nil { 117 117 + ix.indexer = indexer 118 118 + return true, nil 119 119 + } 120 120 + 121 121 + mapping, err := generatePullIndexMapping() 122 122 + if err != nil { 123 123 + return false, err 124 124 + } 125 125 + indexer, err = bleve.New(ix.path, mapping) 126 126 + if err != nil { 127 127 + return false, err 128 128 + } 129 129 + 130 130 + ix.indexer = indexer 131 131 + 132 132 + return false, nil 133 133 + } 134 134 + 135 135 + func openIndexer(ctx context.Context, path string) (bleve.Index, error) { 136 136 + l := tlog.FromContext(ctx) 137 137 + indexer, err := bleve.Open(path) 138 138 + if err != nil { 139 139 + if errors.Is(err, upsidedown.IncompatibleVersion) { 140 140 + l.Info("Indexer was built with a previous version of bleve, deleting and rebuilding") 141 141 + return nil, os.RemoveAll(path) 142 142 + } 143 143 + return nil, nil 144 144 + } 145 145 + return indexer, nil 146 146 + } 147 147 + 148 148 + func PopulateIndexer(ctx context.Context, ix *Indexer, e db.Execer) error { 149 149 + l := tlog.FromContext(ctx) 150 150 + 151 151 + pulls, err := db.GetPulls(e) 152 152 + if err != nil { 153 153 + return err 154 154 + } 155 155 + count := len(pulls) 156 156 + err = ix.Index(ctx, pulls...) 157 157 + if err != nil { 158 158 + return err 159 159 + } 160 160 + l.Info("pulls indexed", "count", count) 161 161 + return err 162 162 + } 163 163 + 164 164 + // pullData data stored and will be indexed 165 165 + type pullData struct { 166 166 + ID int64 `json:"id"` 167 167 + RepoAt string `json:"repo_at"` 168 168 + PullID int `json:"pull_id"` 169 169 + Title string `json:"title"` 170 170 + Body string `json:"body"` 171 171 + State string `json:"state"` 172 172 + 173 173 + Comments []pullCommentData `json:"comments"` 174 174 + } 175 175 + 176 176 + func makePullData(pull *models.Pull) *pullData { 177 177 + return &pullData{ 178 178 + ID: int64(pull.ID), 179 179 + RepoAt: pull.RepoAt.String(), 180 180 + PullID: pull.PullId, 181 181 + Title: pull.Title, 182 182 + Body: pull.Body, 183 183 + State: pull.State.String(), 184 184 + } 185 185 + } 186 186 + 187 187 + // Type returns the document type, for bleve's mapping.Classifier interface. 188 188 + func (i *pullData) Type() string { 189 189 + return pullIndexerDocType 190 190 + } 191 191 + 192 192 + type pullCommentData struct { 193 193 + Body string `json:"body"` 194 194 + } 195 195 + 196 196 + type searchResult struct { 197 197 + Hits []int64 198 198 + Total uint64 199 199 + } 200 200 + 201 201 + const maxBatchSize = 20 202 202 + 203 203 + func (ix *Indexer) Index(ctx context.Context, pulls ...*models.Pull) error { 204 204 + batch := bleveutil.NewFlushingBatch(ix.indexer, maxBatchSize) 205 205 + for _, pull := range pulls { 206 206 + pullData := makePullData(pull) 207 207 + if err := batch.Index(base36.Encode(pullData.ID), pullData); err != nil { 208 208 + return err 209 209 + } 210 210 + } 211 211 + return batch.Flush() 212 212 + } 213 213 + 214 214 + func (ix *Indexer) Delete(ctx context.Context, pullID int64) error { 215 215 + return ix.indexer.Delete(base36.Encode(pullID)) 216 216 + } 217 217 + 218 218 + // Search searches for pulls 219 219 + func (ix *Indexer) Search(ctx context.Context, opts models.PullSearchOptions) (*searchResult, error) { 220 220 + var queries []query.Query 221 221 + 222 222 + // TODO(boltless): remove this after implementing pulls page pagination 223 223 + limit := opts.Page.Limit 224 224 + if limit == 0 { 225 225 + limit = 500 226 226 + } 227 227 + 228 228 + if opts.Keyword != "" { 229 229 + queries = append(queries, bleve.NewDisjunctionQuery( 230 230 + bleveutil.MatchAndQuery("title", opts.Keyword, pullIndexerAnalyzer, 0), 231 231 + bleveutil.MatchAndQuery("body", opts.Keyword, pullIndexerAnalyzer, 0), 232 232 + )) 233 233 + } 234 234 + queries = append(queries, bleveutil.KeywordFieldQuery("repo_at", opts.RepoAt)) 235 235 + queries = append(queries, bleveutil.KeywordFieldQuery("state", opts.State.String())) 236 236 + 237 237 + var indexerQuery query.Query = bleve.NewConjunctionQuery(queries...) 238 238 + searchReq := bleve.NewSearchRequestOptions(indexerQuery, limit, opts.Page.Offset, false) 239 239 + res, err := ix.indexer.SearchInContext(ctx, searchReq) 240 240 + if err != nil { 241 241 + return nil, nil 242 242 + } 243 243 + ret := &searchResult{ 244 244 + Total: res.Total, 245 245 + Hits: make([]int64, len(res.Hits)), 246 246 + } 247 247 + for i, hit := range res.Hits { 248 248 + id, err := base36.Decode(hit.ID) 249 249 + if err != nil { 250 250 + return nil, err 251 251 + } 252 252 + ret.Hits[i] = id 253 253 + } 254 254 + return ret, nil 255 255 + }

+8

appview/models/search.go

··· 10 10 Page pagination.Page 11 11 } 12 12 13 13 + type PullSearchOptions struct { 14 14 + Keyword string 15 15 + RepoAt string 16 16 + State PullState 17 17 + 18 18 + Page pagination.Page 19 19 + } 20 20 + 13 21 // func (so *SearchOptions) ToFilters() []filter { 14 22 // var filters []filter 15 23 // if so.IsOpen != nil {

+3

appview/pulls/pulls.go

··· 2170 2170 s.pages.Notice(w, "pull-merge-error", "Failed to merge pull request. Try again later.") 2171 2171 return 2172 2172 } 2173 2173 + p.State = models.PullMerged 2173 2174 } 2174 2175 2175 2176 err = tx.Commit() ··· 2243 2244 s.pages.Notice(w, "pull-close", "Failed to close pull.") 2244 2245 return 2245 2246 } 2247 2247 + p.State = models.PullClosed 2246 2248 } 2247 2249 2248 2250 // Commit the transaction ··· 2315 2317 s.pages.Notice(w, "pull-close", "Failed to close pull.") 2316 2318 return 2317 2319 } 2320 2320 + p.State = models.PullOpen 2318 2321 } 2319 2322 2320 2323 // Commit the transaction