服务端(Go): - GM 塞背包/带属性装备发放:在线角色推送 op14 增量物品行,背包实时刷新,无需回角色选择界面 - GM 改金币/改等级:去掉在线拒绝,落库后推送 op14 钱包/op37 等级状态,在线实时刷新 - 等级端点统一切换到新版 GMSetCharacterLevel(自带在线刷新) - 无限/锁死疲劳模式 + 燃烧疲劳:每房间消耗 2 点、打怪与通关经验双倍 - 燃烧疲劳开关时给所有在线角色推送系统提示(op12 mode 0) - 深渊史诗掉率:GM 全局倍率 + 城镇增益药剂窗口倍率,真实接入掉落结算 启动器(Python/JS): - 标题栏自绘、窗口 1204x717、禁止拖拽改大小、仅最小化/关闭 - GM 工具:疲劳与燃烧、深渊史诗掉率、拍卖行上架(单一定价对齐) - 拍卖行上架前端对齐服务端单一价格语义,修复 400 报错
819 lines
25 KiB
Go
819 lines
25 KiB
Go
package services
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"errors"
|
|
"fmt"
|
|
"log"
|
|
"net/url"
|
|
"os"
|
|
"path/filepath"
|
|
"regexp"
|
|
"runtime"
|
|
"strings"
|
|
"sync"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"pvfine/internal/pvf"
|
|
)
|
|
|
|
const advancedQueryBudget int64 = 256 << 20
|
|
const advancedQueryLimit = 8
|
|
const advancedCursorStride uint64 = 1 << 32
|
|
|
|
var advancedQuerySequence atomic.Uint64
|
|
var ErrAdvancedCursorStale = errors.New("高级搜索游标已失效,请重新搜索")
|
|
|
|
type advancedQueryKey struct {
|
|
query, scope string
|
|
regex bool
|
|
}
|
|
type advancedDiskQuery struct {
|
|
id uint64
|
|
key advancedQueryKey
|
|
path string
|
|
count int
|
|
bytes int64
|
|
used uint64
|
|
nameIndexReady bool
|
|
semanticGeneration uint64
|
|
}
|
|
|
|
// A session owns immutable reference shards, a metadata/override projection,
|
|
// and at most eight disk result sets.
|
|
// mu serializes operations, but is NEVER acquired while holding core.mu.
|
|
// Edits cancel result generation and retain completed shards for delta updates;
|
|
// explicit cancellation and closing dispose of the session asynchronously.
|
|
// This lets close/edit obtain core.mu between scanner steps instead of waiting
|
|
// for a whole archive scan or SQLite sort.
|
|
type advancedSQLite struct {
|
|
cacheReleases []func()
|
|
mu sync.Mutex
|
|
ctx context.Context
|
|
cancel context.CancelFunc
|
|
dir string
|
|
cachePath string
|
|
queryBudget int64 // zero uses the default; injectable for storage-failure tests
|
|
db *sql.DB
|
|
shards int
|
|
ready bool
|
|
queries map[uint64]*advancedDiskQuery
|
|
clock uint64
|
|
closed chan struct{}
|
|
closeOnce sync.Once
|
|
published atomic.Bool
|
|
epoch uint64 // core.mu protects epoch, pendingIDs and activeCancel
|
|
pendingIDs map[uint64]struct{}
|
|
activeCancel context.CancelFunc
|
|
requestCtx context.Context // d.mu protects request context and projection state
|
|
indexedEpoch uint64
|
|
entries []uint64
|
|
poolState pvf.StringPoolState
|
|
overlay bool
|
|
mutable bool
|
|
}
|
|
|
|
func (d *advancedSQLite) close() {
|
|
d.closeOnce.Do(func() {
|
|
d.cancel()
|
|
d.mu.Lock()
|
|
defer d.mu.Unlock()
|
|
if d.db != nil {
|
|
_ = d.db.Close()
|
|
d.db = nil
|
|
}
|
|
if d.dir != "" {
|
|
_ = os.RemoveAll(d.dir)
|
|
}
|
|
for _, release := range d.cacheReleases {
|
|
release()
|
|
}
|
|
d.cacheReleases = nil
|
|
close(d.closed)
|
|
})
|
|
}
|
|
|
|
// CancelAdvancedSearch releases the lazy string index and cancels its current
|
|
// build/query. Binary search keeps its existing API and cancellation behavior.
|
|
func (s *ArchiveService) CancelAdvancedSearch() {
|
|
s.c.mu.Lock()
|
|
s.c.invalidateAdvancedSearchLocked()
|
|
s.c.mu.Unlock()
|
|
emitEvent("archive:advanced-search-stale")
|
|
}
|
|
|
|
func openAdvancedDB(path string) (*sql.DB, error) {
|
|
// All PRAGMAs are connection-local; the single-connection pool ensures they
|
|
// apply to every operation. Results/index have separate bounded connections.
|
|
db, err := sql.Open("sqlite", path)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
db.SetMaxOpenConns(1)
|
|
db.SetMaxIdleConns(1)
|
|
_, err = db.Exec(`PRAGMA journal_mode=OFF; PRAGMA synchronous=OFF; PRAGMA cache_size=-8192; PRAGMA temp_store=FILE; PRAGMA mmap_size=0; PRAGMA busy_timeout=5000`)
|
|
if err != nil {
|
|
db.Close()
|
|
return nil, err
|
|
}
|
|
return db, nil
|
|
}
|
|
|
|
func (s *ArchiveService) searchAdvancedStringSQLite(query, scope string, regex bool, cursor, limit int) (*AdvancedSearchResult, error) {
|
|
var match func(string) bool
|
|
if regex {
|
|
re, err := regexp.Compile(query)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("正则表达式无效: %w", err)
|
|
}
|
|
match = re.MatchString
|
|
} else {
|
|
lower := strings.ToLower(query)
|
|
match = func(value string) bool { return strings.Contains(strings.ToLower(value), lower) }
|
|
}
|
|
c := s.c
|
|
finishTask := c.archiveTasks.begin()
|
|
defer finishTask()
|
|
c.mu.Lock()
|
|
if c.archive == nil {
|
|
c.mu.Unlock()
|
|
return nil, ErrNoArchive
|
|
}
|
|
a := c.archive
|
|
d := c.advancedDisk
|
|
if d == nil {
|
|
if cursor != 0 {
|
|
c.mu.Unlock()
|
|
return nil, ErrAdvancedCursorStale
|
|
}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
d = &advancedSQLite{ctx: ctx, cancel: cancel, queries: make(map[uint64]*advancedDiskQuery), closed: make(chan struct{})}
|
|
c.advancedDisk = d
|
|
c.advancedCancel = cancel
|
|
}
|
|
c.mu.Unlock()
|
|
d.mu.Lock()
|
|
defer d.mu.Unlock()
|
|
c.mu.Lock()
|
|
if c.advancedDisk != d || c.archive != a {
|
|
c.mu.Unlock()
|
|
return nil, ErrAdvancedCursorStale
|
|
}
|
|
requestCtx, requestCancel := context.WithCancel(d.ctx)
|
|
d.requestCtx, d.activeCancel = requestCtx, requestCancel
|
|
epoch := d.epoch
|
|
c.mu.Unlock()
|
|
defer func() { requestCancel(); c.mu.Lock(); d.activeCancel = nil; c.mu.Unlock(); d.requestCtx = nil }()
|
|
if d.ready && epoch != d.indexedEpoch {
|
|
if cursor != 0 {
|
|
return nil, ErrAdvancedCursorStale
|
|
}
|
|
if err := d.syncDelta(c, a, epoch); err != nil {
|
|
if errors.Is(err, errAdvancedPoolChanged) {
|
|
c.mu.Lock()
|
|
if c.advancedDisk == d {
|
|
c.invalidateAdvancedSearchLocked()
|
|
}
|
|
c.mu.Unlock()
|
|
return nil, errAdvancedPoolChanged
|
|
}
|
|
if !errors.Is(err, context.Canceled) && !errors.Is(err, ErrAdvancedCursorStale) {
|
|
d.status(c, AdvancedSearchIndexStatus{State: AdvancedIndexStateError, Stage: "delta-error", Error: err.Error()})
|
|
}
|
|
if d.db == nil {
|
|
c.mu.Lock()
|
|
if c.advancedDisk == d {
|
|
c.invalidateAdvancedSearchLocked()
|
|
}
|
|
c.mu.Unlock()
|
|
}
|
|
return nil, err
|
|
}
|
|
}
|
|
if err := d.ctx.Err(); err != nil {
|
|
return nil, ErrAdvancedCursorStale
|
|
}
|
|
if !d.ready {
|
|
if err := d.build(c, a); err != nil {
|
|
d.status(c, AdvancedSearchIndexStatus{State: AdvancedIndexStateError, Stage: "error", Error: err.Error()})
|
|
// Failed candidates are never usable; the next first-page request retries.
|
|
c.mu.Lock()
|
|
if c.advancedDisk == d {
|
|
c.invalidateAdvancedSearchLocked()
|
|
if !errors.Is(err, context.Canceled) {
|
|
c.advancedStatus = AdvancedSearchIndexStatus{State: AdvancedIndexStateError, Stage: "error", Error: err.Error()}
|
|
}
|
|
}
|
|
c.mu.Unlock()
|
|
return nil, err
|
|
}
|
|
}
|
|
d.publishCurrentCache(c, a)
|
|
key := advancedQueryKey{query: query, scope: scope, regex: regex}
|
|
var q *advancedDiskQuery
|
|
after := 0
|
|
if cursor != 0 {
|
|
id := uint64(cursor) / advancedCursorStride
|
|
after = int(uint64(cursor) % advancedCursorStride)
|
|
q = d.queries[id]
|
|
if q == nil || q.key != key || after > q.count {
|
|
return nil, ErrAdvancedCursorStale
|
|
}
|
|
c.mu.RLock()
|
|
namesReady := c.indexStatus.State == IndexStateReady && (c.diskIndex == nil || !c.diskIndex.dirty)
|
|
semanticGeneration := c.indexGen
|
|
c.mu.RUnlock()
|
|
if q.semanticGeneration != semanticGeneration || q.nameIndexReady != namesReady {
|
|
return nil, ErrAdvancedCursorStale
|
|
}
|
|
} else {
|
|
for _, existing := range d.queries {
|
|
if existing.key == key {
|
|
c.mu.RLock()
|
|
nameIndexReady := c.indexStatus.State == IndexStateReady && (c.diskIndex == nil || !c.diskIndex.dirty)
|
|
semanticGeneration := c.indexGen
|
|
c.mu.RUnlock()
|
|
if existing.semanticGeneration != semanticGeneration || nameIndexReady != existing.nameIndexReady {
|
|
delete(d.queries, existing.id)
|
|
_ = os.Remove(existing.path)
|
|
continue
|
|
}
|
|
q = existing
|
|
break
|
|
}
|
|
}
|
|
if q == nil {
|
|
var err error
|
|
q, err = d.query(c, key, match)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
}
|
|
d.clock++
|
|
q.used = d.clock
|
|
result, err := d.page(q, after, limit)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
c.mu.RLock()
|
|
defer c.mu.RUnlock()
|
|
if c.archive != a || c.advancedDisk != d || d.requestContext().Err() != nil || d.epoch != d.indexedEpoch {
|
|
return nil, ErrAdvancedCursorStale
|
|
}
|
|
for _, hit := range result.Hits {
|
|
hit.Name = c.indexedFileNameLocked(hit.FileIndex)
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
func (d *advancedSQLite) status(c *core, status AdvancedSearchIndexStatus) {
|
|
c.mu.Lock()
|
|
current := c.advancedDisk == d && d.ctx.Err() == nil
|
|
if current {
|
|
c.advancedStatus = status
|
|
}
|
|
c.mu.Unlock()
|
|
if !current {
|
|
return
|
|
}
|
|
event := "archive:advanced-index-progress"
|
|
if status.State == AdvancedIndexStateReady {
|
|
event = "archive:advanced-index-ready"
|
|
}
|
|
if status.State == AdvancedIndexStateError {
|
|
event = "archive:advanced-index-error"
|
|
}
|
|
emitEvent(event, status)
|
|
}
|
|
|
|
func (d *advancedSQLite) build(c *core, a *pvf.Archive) error {
|
|
started := time.Now()
|
|
root, err := os.UserCacheDir()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
root = filepath.Join(root, "pvfine", "advanced-search")
|
|
finishIO := holdCachePaths(root)
|
|
defer finishIO()
|
|
if err = os.MkdirAll(root, 0700); err != nil {
|
|
return err
|
|
}
|
|
d.dir, err = os.MkdirTemp(root, "session-")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
d.cacheReleases = append(d.cacheReleases, holdCachePaths(d.dir))
|
|
c.mu.RLock()
|
|
if c.advancedDisk != d || c.archive != a || d.ctx.Err() != nil {
|
|
c.mu.RUnlock()
|
|
return context.Canceled
|
|
}
|
|
digest, clean := a.SavedContentHash()
|
|
totalFiles := int(a.FileCount())
|
|
c.mu.RUnlock()
|
|
if clean {
|
|
// Schema/rule version is part of the filename; modified archives only use
|
|
// session files and can never be published as a clean cached index.
|
|
d.cachePath = filepath.Join(root, "refs-v4-"+digest)
|
|
d.cacheReleases = append(d.cacheReleases, holdCachePaths(d.cachePath))
|
|
if d.restoreCache() {
|
|
d.ready = true
|
|
if err := d.captureBaseline(c, a); err != nil {
|
|
return err
|
|
}
|
|
d.status(c, AdvancedSearchIndexStatus{State: AdvancedIndexStateReady, Stage: "ready-cache", Done: totalFiles, Total: totalFiles})
|
|
developmentLog("[pvfine:advanced] sqlite reverse index cache hit: files=%d elapsed=%s", totalFiles, time.Since(started).Round(time.Millisecond))
|
|
return nil
|
|
}
|
|
}
|
|
d.db, err = openAdvancedDB(filepath.Join(d.dir, "index.db"))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = d.db.ExecContext(d.ctx, `CREATE TABLE meta(version INTEGER,shards INTEGER); CREATE TABLE pool(offset INTEGER PRIMARY KEY,pool TEXT NOT NULL,value TEXT NOT NULL,resolved TEXT NOT NULL);
|
|
CREATE TABLE files(file_index INTEGER PRIMARY KEY,path TEXT NOT NULL,lower_path TEXT NOT NULL,size INTEGER,data_type INTEGER);`)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
c.mu.RLock()
|
|
if c.advancedDisk != d || c.archive != a || d.ctx.Err() != nil {
|
|
c.mu.RUnlock()
|
|
return context.Canceled
|
|
}
|
|
total := int(a.FileCount())
|
|
pools := a.NewStringPoolScanner()
|
|
scanner := a.NewStringReferenceScanner()
|
|
c.mu.RUnlock()
|
|
d.status(c, AdvancedSearchIndexStatus{State: AdvancedIndexStateBuilding, Stage: "strings", Total: total})
|
|
developmentLog("[pvfine:advanced] sqlite build started: files=%d", total)
|
|
// Fixed-size transactions keep SQLite's pending writes bounded. Archive read
|
|
// locks cover decoding only, never database writes or index creation.
|
|
tx, err := d.db.BeginTx(d.ctx, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer func() {
|
|
if tx != nil {
|
|
_ = tx.Rollback()
|
|
}
|
|
}()
|
|
entries, refs, files, pending := 0, 0, 0, 0
|
|
var poolStmt, fileStmt *sql.Stmt
|
|
prepare := func() error {
|
|
var e error
|
|
poolStmt, e = tx.PrepareContext(d.ctx, `INSERT INTO pool VALUES(?,?,?,?)`)
|
|
if e != nil {
|
|
return e
|
|
}
|
|
fileStmt, e = tx.PrepareContext(d.ctx, `INSERT INTO files VALUES(?,?,?,?,?)`)
|
|
return e
|
|
}
|
|
closeStatements := func() {
|
|
if poolStmt != nil {
|
|
poolStmt.Close()
|
|
}
|
|
if fileStmt != nil {
|
|
fileStmt.Close()
|
|
}
|
|
}
|
|
defer closeStatements()
|
|
if err = prepare(); err != nil {
|
|
return err
|
|
}
|
|
flush := func() error {
|
|
if pending < 4096 {
|
|
return nil
|
|
}
|
|
closeStatements()
|
|
if e := tx.Commit(); e != nil {
|
|
return e
|
|
}
|
|
var e error
|
|
tx, e = d.db.BeginTx(d.ctx, nil)
|
|
if e != nil {
|
|
return e
|
|
}
|
|
pending = 0
|
|
return prepare()
|
|
}
|
|
for {
|
|
c.mu.RLock()
|
|
if c.advancedDisk != d || c.archive != a || d.ctx.Err() != nil {
|
|
c.mu.RUnlock()
|
|
return context.Canceled
|
|
}
|
|
entry, ok, e := pools.Next(d.ctx)
|
|
resolved := ""
|
|
if ok && e == nil && strings.Contains(entry.Value, "<") && strings.Contains(entry.Value, "::") {
|
|
resolved = a.ResolvePlaceholders(entry.Value)
|
|
if resolved == entry.Value {
|
|
resolved = ""
|
|
}
|
|
}
|
|
c.mu.RUnlock()
|
|
if e != nil {
|
|
return e
|
|
}
|
|
if !ok {
|
|
break
|
|
}
|
|
if _, e = poolStmt.ExecContext(d.ctx, entry.Offset, entry.Pool, entry.Value, resolved); e != nil {
|
|
return e
|
|
}
|
|
entries++
|
|
pending++
|
|
if e = flush(); e != nil {
|
|
return e
|
|
}
|
|
}
|
|
developmentLog("[pvfine:advanced] sqlite strings=%d elapsed=%s", entries, time.Since(started).Round(time.Millisecond))
|
|
referencesStarted := time.Now()
|
|
d.status(c, AdvancedSearchIndexStatus{State: AdvancedIndexStateBuilding, Stage: "references", Total: total})
|
|
workers := max(1, min(advancedMaxShards, runtime.GOMAXPROCS(0)-1, total))
|
|
d.shards = workers
|
|
writers := make([]*advancedReferenceWriter, 0, workers)
|
|
defer func() {
|
|
for _, writer := range writers {
|
|
writer.close()
|
|
}
|
|
}()
|
|
for shard := 0; shard < workers; shard++ {
|
|
writer, e := newAdvancedReferenceWriter(d.ctx, d.dir, shard)
|
|
if e != nil {
|
|
return e
|
|
}
|
|
writers = append(writers, writer)
|
|
}
|
|
pipeline := d.startReferencePipeline(c, a, scanner, workers, func(ctx context.Context, shard int, result advancedReferenceResult) error {
|
|
return writers[shard].add(ctx, result.index, result.refs)
|
|
})
|
|
defer pipeline.close()
|
|
for result := range pipeline.results {
|
|
index, path, file := result.index, result.path, result.file
|
|
var e error
|
|
if _, e = fileStmt.ExecContext(d.ctx, index, path, strings.ToLower(path), file.DataSize, file.DataType); e != nil {
|
|
return e
|
|
}
|
|
pending++
|
|
refs += result.refCount
|
|
if e = flush(); e != nil {
|
|
return e
|
|
}
|
|
files++
|
|
if files%16384 == 0 {
|
|
d.status(c, AdvancedSearchIndexStatus{State: AdvancedIndexStateBuilding, Stage: "references", Done: files, Total: total})
|
|
}
|
|
}
|
|
if err = pipeline.err(); err != nil {
|
|
return err
|
|
}
|
|
closeStatements()
|
|
if err = tx.Commit(); err != nil {
|
|
return err
|
|
}
|
|
developmentLog("[pvfine:advanced] sqlite references=%d files=%d workers=%d elapsed=%s", refs, files, workers, time.Since(referencesStarted).Round(time.Millisecond))
|
|
d.status(c, AdvancedSearchIndexStatus{State: AdvancedIndexStateBuilding, Stage: "indexes", Done: files, Total: total})
|
|
if err = finishAdvancedShards(d.ctx, writers); err != nil {
|
|
return err
|
|
}
|
|
for _, writer := range writers {
|
|
writer.close()
|
|
}
|
|
writers = nil
|
|
_, err = d.db.ExecContext(d.ctx, `CREATE INDEX files_path ON files(path,file_index);`)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if _, err = d.db.ExecContext(d.ctx, `INSERT INTO meta VALUES(4,?)`, d.shards); err != nil {
|
|
return err
|
|
}
|
|
if err = d.db.Close(); err != nil {
|
|
return err
|
|
}
|
|
d.db, err = openAdvancedReadDB(filepath.Join(d.dir, "index.db"))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if d.cachePath != "" && d.ctx.Err() == nil {
|
|
if e := d.publishCache(root); e != nil && !os.IsExist(e) {
|
|
log.Printf("[pvfine:advanced] cache publication unavailable; using session index")
|
|
}
|
|
requestCacheMaintenance()
|
|
}
|
|
d.ready = true
|
|
if err := d.captureBaseline(c, a); err != nil {
|
|
return err
|
|
}
|
|
d.status(c, AdvancedSearchIndexStatus{State: AdvancedIndexStateReady, Stage: "ready-sqlite", Done: files, Total: total})
|
|
size := advancedIndexDirectorySize(d.dir)
|
|
developmentLog("[pvfine:advanced] sqlite build finished: strings=%d references=%d files=%d bytes=%d elapsed=%s", entries, refs, files, size, time.Since(started).Round(time.Millisecond))
|
|
return nil
|
|
}
|
|
|
|
func attachAdvancedSource(ctx context.Context, db *sql.DB, path string) error {
|
|
_, err := db.ExecContext(ctx, `ATTACH DATABASE ? AS source`, sqliteFileURI(path, "mode=ro"))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = db.ExecContext(ctx, `PRAGMA source.cache_size=-8192; PRAGMA source.mmap_size=0`)
|
|
return err
|
|
}
|
|
|
|
func sqliteFileURI(path, query string) string {
|
|
path = filepath.ToSlash(filepath.Clean(path))
|
|
u := url.URL{Scheme: "file", RawQuery: query}
|
|
if strings.HasPrefix(path, "//") {
|
|
rest := strings.TrimPrefix(path, "//")
|
|
if slash := strings.IndexByte(rest, '/'); slash >= 0 {
|
|
u.Host = rest[:slash]
|
|
u.Path = rest[slash:]
|
|
} else {
|
|
u.Host = rest
|
|
u.Path = "/"
|
|
}
|
|
} else {
|
|
if len(path) >= 2 && path[1] == ':' {
|
|
path = "/" + path
|
|
}
|
|
if !strings.HasPrefix(path, "/") {
|
|
path = "/" + path
|
|
}
|
|
u.Path = path
|
|
}
|
|
return u.String()
|
|
}
|
|
|
|
func (d *advancedSQLite) evictOldest() error {
|
|
var oldest *advancedDiskQuery
|
|
for _, q := range d.queries {
|
|
if oldest == nil || q.used < oldest.used {
|
|
oldest = q
|
|
}
|
|
}
|
|
if oldest == nil {
|
|
return nil
|
|
}
|
|
if err := os.Remove(oldest.path); err != nil && !os.IsNotExist(err) {
|
|
return err
|
|
}
|
|
delete(d.queries, oldest.id)
|
|
return nil
|
|
}
|
|
|
|
func (d *advancedSQLite) query(c *core, key advancedQueryKey, match func(string) bool) (*advancedDiskQuery, error) {
|
|
started := time.Now()
|
|
budget := d.queryBudget
|
|
if budget == 0 {
|
|
budget = advancedQueryBudget
|
|
}
|
|
remaining := func() int64 {
|
|
n := budget
|
|
for _, q := range d.queries {
|
|
n -= q.bytes
|
|
}
|
|
return n
|
|
}
|
|
// Reserve at least 64 MiB before starting. max_page_count makes an oversized
|
|
// current query fail explicitly instead of silently exceeding the disk budget.
|
|
for len(d.queries) >= advancedQueryLimit || remaining() < min(64<<20, budget) {
|
|
if err := d.evictOldest(); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
id := advancedQuerySequence.Add(1)
|
|
// Cursors round-trip through JavaScript numbers without precision loss.
|
|
if id >= 1<<21 {
|
|
return nil, errors.New("高级搜索会话编号已耗尽,请重启应用")
|
|
}
|
|
q := &advancedDiskQuery{id: id, key: key, path: filepath.Join(d.dir, fmt.Sprintf("query-%d.db", id))}
|
|
c.mu.RLock()
|
|
q.nameIndexReady = c.indexStatus.State == IndexStateReady && (c.diskIndex == nil || !c.diskIndex.dirty)
|
|
q.semanticGeneration = c.indexGen
|
|
c.mu.RUnlock()
|
|
db, err := openAdvancedDB(q.path)
|
|
if err != nil {
|
|
_ = os.Remove(q.path)
|
|
return nil, err
|
|
}
|
|
success := false
|
|
defer func() {
|
|
db.Close()
|
|
if !success {
|
|
_ = os.Remove(q.path)
|
|
}
|
|
}()
|
|
var pageSize int64
|
|
if err = db.QueryRowContext(d.requestContext(), `PRAGMA page_size`).Scan(&pageSize); err != nil {
|
|
return nil, err
|
|
}
|
|
if _, err = db.ExecContext(d.requestContext(), fmt.Sprintf("PRAGMA max_page_count=%d", remaining()/pageSize)); err != nil {
|
|
return nil, err
|
|
}
|
|
if err = attachAdvancedSource(d.requestContext(), db, filepath.Join(d.dir, "index.db")); err != nil {
|
|
return nil, err
|
|
}
|
|
if err = d.attachShards(d.requestContext(), db); err != nil {
|
|
return nil, err
|
|
}
|
|
if _, err = db.ExecContext(d.requestContext(), `CREATE TABLE matches(offset INTEGER PRIMARY KEY); CREATE TABLE name_matches(file_index INTEGER NOT NULL,value TEXT NOT NULL,PRIMARY KEY(file_index,value)) WITHOUT ROWID; CREATE TABLE hits(seq INTEGER PRIMARY KEY,file_index INTEGER NOT NULL);`); err != nil {
|
|
return nil, err
|
|
}
|
|
tx, err := db.BeginTx(d.requestContext(), nil)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer tx.Rollback()
|
|
stmt, err := tx.PrepareContext(d.requestContext(), `INSERT INTO matches VALUES(?)`)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer stmt.Close()
|
|
rows, err := d.db.QueryContext(d.requestContext(), `SELECT offset,value,resolved FROM pool ORDER BY offset`)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
scanned := 0
|
|
for rows.Next() {
|
|
var offset int32
|
|
var value string
|
|
var resolved string
|
|
if err = rows.Scan(&offset, &value, &resolved); err != nil {
|
|
return nil, err
|
|
}
|
|
scanned++
|
|
if match(value) || (resolved != "" && match(resolved)) {
|
|
if _, err = stmt.ExecContext(d.requestContext(), offset); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
}
|
|
if err = rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
rows.Close()
|
|
stmt.Close()
|
|
if err = tx.Commit(); err != nil {
|
|
return nil, err
|
|
}
|
|
if err = d.addAdvancedNameMatches(c, db, key, match); err != nil {
|
|
return nil, err
|
|
}
|
|
// Materialize only ordered file identities. Details remain normalized in the
|
|
// reverse index and are fetched for the current page, without duplicating text.
|
|
var candidates []string
|
|
for shard := 0; shard < d.shards; shard++ {
|
|
query := fmt.Sprintf(`SELECT r.file_index FROM matches m CROSS JOIN refs%d.refs r INDEXED BY refs_offset ON r.offset=m.offset`, shard)
|
|
if d.overlay {
|
|
query = fmt.Sprintf(`SELECT b.file_index FROM matches m CROSS JOIN refs%d.refs r INDEXED BY refs_offset ON r.offset=m.offset JOIN source.file_map b ON b.orig_index=r.file_index AND b.overridden=0`, shard)
|
|
}
|
|
candidates = append(candidates, query)
|
|
}
|
|
if d.overlay {
|
|
candidates = append(candidates, `SELECT r.file_index FROM matches m CROSS JOIN source.delta_refs r INDEXED BY delta_refs_offset ON r.offset=m.offset`)
|
|
}
|
|
candidates = append(candidates, `SELECT file_index FROM name_matches`)
|
|
_, err = db.ExecContext(d.requestContext(), `INSERT INTO hits(file_index)
|
|
SELECT r.file_index FROM (`+strings.Join(candidates, " UNION ALL ")+`) r
|
|
JOIN source.files f ON f.file_index=r.file_index
|
|
WHERE ?='' OR f.lower_path=? OR substr(f.lower_path,1,length(?)+1)=?||'/'
|
|
GROUP BY r.file_index ORDER BY f.path,r.file_index`, key.scope, key.scope, key.scope, key.scope)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("高级搜索结果写入失败(结果磁盘预算 256 MiB,可缩小搜索范围): %w", err)
|
|
}
|
|
if err = db.QueryRowContext(d.requestContext(), `SELECT count(*) FROM hits`).Scan(&q.count); err != nil {
|
|
return nil, err
|
|
}
|
|
info, err := os.Stat(q.path)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
q.bytes = info.Size()
|
|
d.queries[id] = q
|
|
success = true
|
|
developmentLog("[pvfine:advanced] sqlite query finished: strings_scanned=%d files=%d bytes=%d elapsed=%s", scanned, q.count, q.bytes, time.Since(started).Round(time.Millisecond))
|
|
return q, nil
|
|
}
|
|
|
|
func (d *advancedSQLite) page(q *advancedDiskQuery, after, limit int) (*AdvancedSearchResult, error) {
|
|
started := time.Now()
|
|
db, err := openAdvancedDB(q.path)
|
|
if err != nil {
|
|
_ = os.Remove(q.path)
|
|
return nil, err
|
|
}
|
|
defer db.Close()
|
|
if err = attachAdvancedSource(d.requestContext(), db, filepath.Join(d.dir, "index.db")); err != nil {
|
|
return nil, err
|
|
}
|
|
if err = d.attachShards(d.requestContext(), db); err != nil {
|
|
return nil, err
|
|
}
|
|
rows, err := db.QueryContext(d.requestContext(), `SELECT h.seq,f.file_index,f.path,f.size,f.data_type FROM hits h JOIN source.files f ON f.file_index=h.file_index WHERE h.seq>? ORDER BY h.seq LIMIT ?`, after, limit)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
result := &AdvancedSearchResult{Hits: []*AdvancedSearchHit{}, NextCursor: -1}
|
|
last := after
|
|
for rows.Next() {
|
|
hit := &AdvancedSearchHit{}
|
|
if err = rows.Scan(&last, &hit.FileIndex, &hit.Path, &hit.Size, &hit.DataType); err != nil {
|
|
rows.Close()
|
|
return nil, err
|
|
}
|
|
result.Hits = append(result.Hits, hit)
|
|
}
|
|
err = rows.Err()
|
|
rows.Close()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
for _, hit := range result.Hits {
|
|
detailsSQL := fmt.Sprintf(`SELECT p.pool,p.offset,CASE WHEN p.resolved<>'' THEN p.resolved ELSE p.value END,r.occurrences,r.types,r.fields FROM refs%d.refs r JOIN matches m ON m.offset=r.offset JOIN source.pool p ON p.offset=r.offset WHERE r.file_index=? ORDER BY r.offset`, int(hit.FileIndex)%d.shards)
|
|
queryIndex := hit.FileIndex
|
|
if d.overlay {
|
|
var original int32
|
|
var overridden int
|
|
err := db.QueryRowContext(d.requestContext(), `SELECT orig_index,overridden FROM source.file_map WHERE file_index=?`, hit.FileIndex).Scan(&original, &overridden)
|
|
if errors.Is(err, sql.ErrNoRows) || err == nil && overridden != 0 {
|
|
detailsSQL = `SELECT p.pool,p.offset,CASE WHEN p.resolved<>'' THEN p.resolved ELSE p.value END,r.occurrences,r.types,r.fields FROM source.delta_refs r JOIN matches m ON m.offset=r.offset JOIN source.pool p ON p.offset=r.offset WHERE r.file_index=? ORDER BY r.offset`
|
|
} else if err != nil {
|
|
return nil, err
|
|
} else {
|
|
queryIndex = original
|
|
detailsSQL = fmt.Sprintf(`SELECT p.pool,p.offset,CASE WHEN p.resolved<>'' THEN p.resolved ELSE p.value END,r.occurrences,r.types,r.fields FROM refs%d.refs r JOIN matches m ON m.offset=r.offset JOIN source.pool p ON p.offset=r.offset WHERE r.file_index=? ORDER BY r.offset`, int(original)%d.shards)
|
|
}
|
|
}
|
|
details, e := db.QueryContext(d.requestContext(), detailsSQL, queryIndex)
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
for details.Next() {
|
|
detail := &AdvancedSearchDetail{Kind: "string"}
|
|
var types, fields int
|
|
if e = details.Scan(&detail.Pool, &detail.PoolOffset, &detail.Value, &detail.Occurrences, &types, &fields); e != nil {
|
|
details.Close()
|
|
return nil, e
|
|
}
|
|
for types != 0 {
|
|
detail.TokenTypes = append(detail.TokenTypes, int32(types&15))
|
|
types >>= 4
|
|
}
|
|
if fields&1 != 0 {
|
|
detail.FileFields = append(detail.FileFields, "name")
|
|
}
|
|
if fields&2 != 0 {
|
|
detail.FileFields = append(detail.FileFields, "path")
|
|
}
|
|
hit.Details = append(hit.Details, detail)
|
|
}
|
|
e = details.Err()
|
|
details.Close()
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
nameRows, e := db.QueryContext(d.requestContext(), `SELECT value FROM name_matches WHERE file_index=? ORDER BY value`, hit.FileIndex)
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
for nameRows.Next() {
|
|
detail := &AdvancedSearchDetail{Kind: "name", Occurrences: 1}
|
|
if e = nameRows.Scan(&detail.Value); e != nil {
|
|
nameRows.Close()
|
|
return nil, e
|
|
}
|
|
hit.Details = append(hit.Details, detail)
|
|
}
|
|
e = nameRows.Err()
|
|
nameRows.Close()
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
}
|
|
if last < q.count {
|
|
result.NextCursor = int(q.id*advancedCursorStride + uint64(last))
|
|
}
|
|
result.Scanned = last
|
|
developmentLog("[pvfine:advanced] sqlite page: hits=%d position=%d total=%d elapsed=%s", len(result.Hits), last, q.count, time.Since(started).Round(time.Millisecond))
|
|
return result, nil
|
|
}
|
|
|
|
func openAdvancedReadDB(path string) (*sql.DB, error) {
|
|
db, err := sql.Open("sqlite", sqliteFileURI(path, "mode=ro"))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
db.SetMaxOpenConns(1)
|
|
db.SetMaxIdleConns(1)
|
|
if _, err = db.Exec(`PRAGMA cache_size=-8192; PRAGMA temp_store=FILE; PRAGMA mmap_size=0`); err != nil {
|
|
db.Close()
|
|
return nil, err
|
|
}
|
|
return db, nil
|
|
}
|