-
Notifications
You must be signed in to change notification settings - Fork 206
/
search.go
468 lines (414 loc) · 14.6 KB
/
search.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
// Copyright 2016 Documize Inc. <legal@documize.com>. All rights reserved.
//
// This software (Documize Community Edition) is licensed under
// GNU AGPL v3 http://www.gnu.org/licenses/agpl-3.0.en.html
//
// You can operate outside the AGPL restrictions by purchasing
// Documize Enterprise Edition and obtaining a commercial license
// by contacting <sales@documize.com>.
//
// https://documize.com
package request
import (
"errors"
"fmt"
"strings"
"sync"
"time"
"github.com/documize/community/core/api/entity"
"github.com/documize/community/core/log"
"github.com/documize/community/core/streamutil"
"github.com/documize/community/core/stringutil"
_ "github.com/go-sql-driver/mysql" // required for sqlx but not directly called
"github.com/jmoiron/sqlx"
)
// SearchManager type provides the datastructure for the queues of activity to be serialized through a single background goroutine.
// NOTE if the queue becomes full, the system will trigger the rebuilding entire files in order to clear the backlog.
type SearchManager struct {
queue chan queueEntry
rebuild map[string]bool
rebuildLock sync.RWMutex
givenWarning bool
}
const searchQueueLength = 2048 // NOTE the largest 15Mb docx in the test set generates 2142 queue entries, but the queue is constantly emptied
type queueEntry struct {
action func(*databaseRequest, entity.Page) error
isRebuild bool
entity.Page
}
func init() {
searches = &SearchManager{}
searches.queue = make(chan queueEntry, searchQueueLength) // provide some decoupling
searches.rebuild = make(map[string]bool)
go searches.searchProcessQueue()
}
// searchProcessQueue is run as a goroutine, it processes the queue of search index update requests.
func (m *SearchManager) searchProcessQueue() {
for {
//fmt.Println("DEBUG queue length=", len(Searches.queue))
if len(m.queue) <= searchQueueLength/20 { // on a busy server, the queue may never get to zero - so use 5%
m.rebuildLock.Lock()
for docid := range m.rebuild {
m.queue <- queueEntry{
action: searchRebuild,
isRebuild: true,
Page: entity.Page{DocumentID: docid},
}
delete(m.rebuild, docid)
}
m.rebuildLock.Unlock()
}
qe := <-m.queue
doit := true
if len(qe.DocumentID) > 0 {
m.rebuildLock.RLock()
if m.rebuild[qe.DocumentID] {
doit = false // don't execute an action on a document queued to be rebuilt
}
m.rebuildLock.RUnlock()
}
if doit {
tx, err := Db.Beginx()
if err != nil {
log.Error("Search Queue Beginx()", err)
} else {
dbRequest := &databaseRequest{Transaction: tx, OrgID: qe.Page.OrgID}
err = qe.action(dbRequest, qe.Page)
if err != nil {
log.Error("Search Queue action()", err)
log.IfErr(tx.Rollback())
// This action has failed, so re-build indexes for the entire document,
// provided it was not a re-build command that failed and we know the documentId.
if !qe.isRebuild && len(qe.DocumentID) > 0 {
m.rebuildLock.Lock()
m.rebuild[qe.DocumentID] = true
m.rebuildLock.Unlock()
}
} else {
log.IfErr(tx.Commit())
}
}
}
}
}
func (m *SearchManager) addQueue(request *databaseRequest, qe queueEntry) error {
lsq := len(m.queue)
if lsq >= (searchQueueLength - 1) {
if qe.DocumentID != "" {
m.rebuildLock.Lock()
if !m.rebuild[qe.DocumentID] {
log.Info(fmt.Sprintf("WARNING: Search Queue Has No Space! Marked rebuild index for document id %s", qe.DocumentID))
}
m.rebuild[qe.DocumentID] = true
m.rebuildLock.Unlock()
} else {
log.Error("addQueue", errors.New("WARNING: Search Queue Has No Space! But unable to index unknown document id"))
}
return nil
}
if lsq > ((8 * searchQueueLength) / 10) {
if !m.givenWarning {
log.Info(fmt.Sprintf("WARNING: Searches.queue length %d exceeds 80%% of capacity", lsq))
m.givenWarning = true
}
} else {
if m.givenWarning {
log.Info(fmt.Sprintf("INFO: Searches.queue length %d now below 80%% of capacity", lsq))
m.givenWarning = false
}
}
m.queue <- qe
return nil
}
// Add should be called when a new page is added to a document.
func (m *SearchManager) Add(request *databaseRequest, page entity.Page, id string) (err error) {
page.RefID = id
err = m.addQueue(request, queueEntry{
action: searchAdd,
Page: page,
})
return
}
func searchAdd(request *databaseRequest, page entity.Page) (err error) {
id := page.RefID
// translate the html into text for the search
nonHTML, err := stringutil.HTML(page.Body).Text(false)
if err != nil {
log.Error("Unable to decode the html for searching", err)
return
}
// insert into the search table, getting the document title along the way
var stmt *sqlx.Stmt
stmt, err = request.Transaction.Preparex(
"INSERT INTO search (id, orgid, documentid, level, sequence, documenttitle, slug, pagetitle, body, created, revised) " +
" SELECT page.refid,page.orgid,document.refid,page.level,page.sequence,document.title,document.slug,page.title,?,page.created,page.revised " +
" FROM document,page WHERE page.refid=? AND document.refid=page.documentid")
if err != nil {
log.Error("Unable to prepare insert for search", err)
return
}
defer streamutil.Close(stmt)
_, err = stmt.Exec(nonHTML, id)
if err != nil {
log.Error(fmt.Sprintf("Unable to execute insert for search"), err)
return
}
return
}
// Update should be called after a page record has been updated.
func (m *SearchManager) Update(request *databaseRequest, page entity.Page) (err error) {
err = m.addQueue(request, queueEntry{
action: searchUpdate,
Page: page,
})
return
}
func searchUpdate(request *databaseRequest, page entity.Page) (err error) {
// translate the html into text for the search
nonHTML, err := stringutil.HTML(page.Body).Text(false)
if err != nil {
log.Error("Unable to decode the html for searching", err)
return
}
su, err := request.Transaction.Preparex(
"UPDATE search SET pagetitle=?,body=?,sequence=?,level=?,revised=? WHERE id=?")
if err != nil {
log.Error(fmt.Sprintf("Unable to prepare search update for page %s", page.RefID), err)
return err // could have been redefined
}
defer streamutil.Close(su)
_, err = su.Exec(page.Title, nonHTML, page.Sequence, page.Level, page.Revised, page.RefID)
if err != nil {
log.Error(fmt.Sprintf("Unable to execute search update for page %s", page.RefID), err)
return
}
return
}
// UpdateDocument should be called after a document record has been updated.
func (m *SearchManager) UpdateDocument(request *databaseRequest, document entity.Document) (err error) {
err = m.addQueue(request, queueEntry{
action: searchUpdateDocument,
Page: entity.Page{
DocumentID: document.RefID,
Title: document.Title,
Body: document.Slug, // NOTE body==slug in this context
},
})
return
}
func searchUpdateDocument(request *databaseRequest, page entity.Page) (err error) {
searchstmt, err := request.Transaction.Preparex(
"UPDATE search SET documenttitle=?, slug=?, revised=? WHERE documentid=?")
if err != nil {
log.Error(fmt.Sprintf("Unable to prepare search update for document %s", page.DocumentID), err)
return err // may have been redefined
}
defer streamutil.Close(searchstmt)
_, err = searchstmt.Exec(page.Title, page.Body, time.Now().UTC(), page.DocumentID)
if err != nil {
log.Error(fmt.Sprintf("Unable to execute search update for document %s", page.DocumentID), err)
return err
}
return nil
}
// DeleteDocument should be called after a document record has been deleted.
func (m *SearchManager) DeleteDocument(request *databaseRequest, documentID string) (err error) {
if len(documentID) > 0 {
m.queue <- queueEntry{
action: searchDeleteDocument,
Page: entity.Page{DocumentID: documentID},
}
}
return
}
func searchDeleteDocument(request *databaseRequest, page entity.Page) (err error) {
var bm = baseManager{}
_, err = bm.DeleteWhere(request.Transaction,
fmt.Sprintf("DELETE from search WHERE documentid='%s'", page.DocumentID))
if err != nil {
log.Error(fmt.Sprintf("Unable to delete search entries for docId %s", page.DocumentID), err)
}
return
}
func searchRebuild(request *databaseRequest, page entity.Page) (err error) {
log.Info(fmt.Sprintf("SearchRebuild begin for docId %s", page.DocumentID))
start := time.Now()
var bm = baseManager{}
_, err = bm.DeleteWhere(request.Transaction, fmt.Sprintf("DELETE from search WHERE documentid='%s'", page.DocumentID))
if err != nil {
log.Error(fmt.Sprintf("Unable to delete search entries for docId %s prior to rebuild",
page.DocumentID), err)
return err
}
var pages []struct{ ID string }
stmt2, err := request.Transaction.Preparex("SELECT refid as id FROM page WHERE documentid=? ")
if err != nil {
log.Error(fmt.Sprintf("Unable to prepare searchRebuild select for docId %s", page.DocumentID), err)
return err
}
defer streamutil.Close(stmt2)
err = stmt2.Select(&pages, page.DocumentID)
if err != nil {
log.Error(fmt.Sprintf("Unable to execute searchRebuild select for docId %s", page.DocumentID), err)
return err
}
if len(pages) > 0 {
for _, pg := range pages {
err = searchAdd(request, entity.Page{BaseEntity: entity.BaseEntity{RefID: pg.ID}})
if err != nil {
log.Error(fmt.Sprintf("Unable to execute searchAdd from searchRebuild for docId %s pageID %s",
page.DocumentID, pg.ID), err)
return err
}
}
// rebuild doc-level tags & excerpts
// get the 0'th page data and rewrite it
target := entity.Page{}
stmt1, err := request.Transaction.Preparex("SELECT * FROM page WHERE refid=?")
if err != nil {
log.Error(fmt.Sprintf("Unable to prepare select from searchRebuild for pageId %s", pages[0].ID), err)
return err
}
defer streamutil.Close(stmt1)
err = stmt1.Get(&target, pages[0].ID)
if err != nil {
log.Error(fmt.Sprintf("Unable to execute select from searchRebuild for pageId %s", pages[0].ID), err)
return err
}
err = searchUpdate(request, target) // to rebuild the document-level tags + excerpt
if err != nil {
log.Error(fmt.Sprintf("Unable to run searchUpdate in searchRebuild for docId %s", target.DocumentID), err)
return err
}
}
log.Info(fmt.Sprintf("Time to rebuild all search data for documentId %s = %v", page.DocumentID,
time.Since(start)))
return
}
// UpdateSequence should be called after a page record has been resequenced.
func (m *SearchManager) UpdateSequence(request *databaseRequest, documentID, pageID string, sequence float64) (err error) {
err = m.addQueue(request, queueEntry{
action: searchUpdateSequence,
Page: entity.Page{
BaseEntity: entity.BaseEntity{RefID: pageID},
Sequence: sequence,
DocumentID: documentID,
},
})
return
}
func searchUpdateSequence(request *databaseRequest, page entity.Page) (err error) {
supdate, err := request.Transaction.Preparex(
"UPDATE search SET sequence=?,revised=? WHERE id=?")
if err != nil {
log.Error(fmt.Sprintf("Unable to prepare search sequence update for page %s", page.RefID), err)
return err
}
defer streamutil.Close(supdate)
_, err = supdate.Exec(page.Sequence, time.Now().UTC(), page.RefID)
if err != nil {
log.Error(fmt.Sprintf("Unable to execute search sequence update for page %s", page.RefID), err)
return
}
return
}
// UpdateLevel should be called after the level of a page has been changed.
func (m *SearchManager) UpdateLevel(request *databaseRequest, documentID, pageID string, level int) (err error) {
err = m.addQueue(request, queueEntry{
action: searchUpdateLevel,
Page: entity.Page{
BaseEntity: entity.BaseEntity{RefID: pageID},
Level: uint64(level),
DocumentID: documentID,
},
})
return
}
func searchUpdateLevel(request *databaseRequest, page entity.Page) (err error) {
pageID := page.RefID
level := page.Level
supdate, err := request.Transaction.Preparex(
"UPDATE search SET level=?,revised=? WHERE id=?")
if err != nil {
log.Error(fmt.Sprintf("Unable to prepare search level update for page %s", pageID), err)
return err
}
defer streamutil.Close(supdate)
_, err = supdate.Exec(level, time.Now().UTC(), pageID)
if err != nil {
log.Error(fmt.Sprintf("Unable to execute search level update for page %s", pageID), err)
return
}
return
}
// Delete should be called after a page has been deleted.
func (m *SearchManager) Delete(request *databaseRequest, documentID, pageID string) (rows int64, err error) {
err = m.addQueue(request, queueEntry{
action: searchDelete,
Page: entity.Page{
BaseEntity: entity.BaseEntity{RefID: pageID},
DocumentID: documentID,
},
})
return
}
func searchDelete(request *databaseRequest, page entity.Page) (err error) {
var bm = baseManager{}
//_, err = bm.DeleteWhere(request.Transaction, fmt.Sprintf("DELETE FROM search WHERE orgid=\"%s\" AND pageid=\"%s\"", request.OrgID, page.RefID))
_, err = bm.DeleteConstrainedWithID(request.Transaction, "search", request.OrgID, page.RefID)
return
}
/******************
* Sort Page Context
*******************/
// GetPageContext is called to get the context of a page in terms of an headings hierarchy.
func (m *SearchManager) GetPageContext(request *databaseRequest, pageID string, existingContext []string) ([]string, error) {
err := request.MakeTx()
if err != nil {
return nil, err
}
target := entity.Search{}
stmt1, err := request.Transaction.Preparex("SELECT * FROM search WHERE id=?")
if err != nil {
log.Error(fmt.Sprintf("Unable to prepare setPageContext select for pageId %s", pageID), err)
return nil, err
}
defer streamutil.Close(stmt1)
err = stmt1.Get(&target, pageID)
if err != nil {
return existingContext, nil
}
context := append([]string{target.PageTitle}, existingContext...)
if target.Level > 1 { // more levels to process
var next struct{ ID string }
// process the lower levels
stmt2, err := request.Transaction.Preparex("SELECT id FROM search WHERE documentid=? " +
"AND sequence=(SELECT max(sequence) FROM search " +
"WHERE documentid=? AND sequence<? AND level=?)")
if err != nil {
log.Error(fmt.Sprintf("Unable to prepare GetPageContext next select for pageId %s", pageID), err)
return nil, err
}
defer streamutil.Close(stmt2)
err = stmt2.Get(&next, target.DocumentID, target.DocumentID, target.Sequence, target.Level-1)
if err != nil {
if strings.Contains(err.Error(), "no rows in result set") {
return context, nil
}
log.Error(fmt.Sprintf("Unable to execute GetPageContext next select for pageId %s", pageID), err)
return nil, err
}
if len(next.ID) > 0 {
context, err = m.GetPageContext(request, next.ID, context)
if err != nil {
log.Error(fmt.Sprintf("Error calling recursive GetPageContext for pageId %s", pageID), err)
return nil, err
}
} else {
err = fmt.Errorf("search.ID<=0 : %s", next.ID)
log.Error(fmt.Sprintf("Unexpected higher level ID in GetPageContext for pageId %s", pageID), err)
return nil, err
}
}
return context, nil
}