Changeset 0.18.0 (#79)
This commit is contained in:
197
server/memory/scanner.go
Normal file
197
server/memory/scanner.go
Normal file
@@ -0,0 +1,197 @@
|
||||
package memory
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.gobha.me/xcaliber/chat-switchboard/database"
|
||||
"git.gobha.me/xcaliber/chat-switchboard/store"
|
||||
)
|
||||
|
||||
// ScannerConfig holds tunable parameters for the background scanner.
|
||||
type ScannerConfig struct {
|
||||
Interval time.Duration // tick interval (default 10m)
|
||||
Concurrency int // max parallel extractions (default 2)
|
||||
}
|
||||
|
||||
// Scanner periodically finds conversations needing extraction.
|
||||
type Scanner struct {
|
||||
extractor *Extractor
|
||||
stores store.Stores
|
||||
cfg ScannerConfig
|
||||
stopCh chan struct{}
|
||||
wg sync.WaitGroup
|
||||
inFlight sync.Map // channelID+userID → true
|
||||
}
|
||||
|
||||
// NewScanner creates a background extraction scanner.
|
||||
func NewScanner(ext *Extractor, stores store.Stores, cfg ScannerConfig) *Scanner {
|
||||
if cfg.Interval <= 0 {
|
||||
cfg.Interval = 10 * time.Minute
|
||||
}
|
||||
if cfg.Concurrency <= 0 {
|
||||
cfg.Concurrency = 2
|
||||
}
|
||||
return &Scanner{
|
||||
extractor: ext,
|
||||
stores: stores,
|
||||
cfg: cfg,
|
||||
stopCh: make(chan struct{}),
|
||||
}
|
||||
}
|
||||
|
||||
// Start begins the background scanning loop.
|
||||
func (s *Scanner) Start() {
|
||||
s.wg.Add(1)
|
||||
go func() {
|
||||
defer s.wg.Done()
|
||||
log.Printf("🧠 memory extraction scanner started (interval=%s, concurrency=%d)",
|
||||
s.cfg.Interval, s.cfg.Concurrency)
|
||||
|
||||
ticker := time.NewTicker(s.cfg.Interval)
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-s.stopCh:
|
||||
log.Println("🧠 memory extraction scanner stopped")
|
||||
return
|
||||
case <-ticker.C:
|
||||
s.scan()
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// Stop gracefully shuts down the scanner.
|
||||
func (s *Scanner) Stop() {
|
||||
close(s.stopCh)
|
||||
s.wg.Wait()
|
||||
}
|
||||
|
||||
// candidate represents a conversation that may need extraction.
|
||||
type candidate struct {
|
||||
ChannelID string
|
||||
UserID string
|
||||
TeamID string
|
||||
}
|
||||
|
||||
func (s *Scanner) scan() {
|
||||
ctx := context.Background()
|
||||
|
||||
// Check global kill switch
|
||||
if s.stores.GlobalConfig != nil {
|
||||
val, err := s.stores.GlobalConfig.Get(ctx, "memory_extraction_enabled")
|
||||
if err != nil || val == nil {
|
||||
return // disabled or not configured
|
||||
}
|
||||
// val is a map; check the "value" key
|
||||
if enabled, ok := val["value"].(bool); !ok || !enabled {
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
candidates, err := s.findCandidates(ctx)
|
||||
if err != nil {
|
||||
log.Printf("⚠ memory scanner: find candidates: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
if len(candidates) == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
// Semaphore for concurrency control
|
||||
sem := make(chan struct{}, s.cfg.Concurrency)
|
||||
var wg sync.WaitGroup
|
||||
|
||||
for _, c := range candidates {
|
||||
key := c.ChannelID + ":" + c.UserID
|
||||
if _, loaded := s.inFlight.LoadOrStore(key, true); loaded {
|
||||
continue // already in progress
|
||||
}
|
||||
|
||||
sem <- struct{}{} // acquire
|
||||
wg.Add(1)
|
||||
|
||||
go func(cand candidate) {
|
||||
defer wg.Done()
|
||||
defer func() { <-sem }() // release
|
||||
defer s.inFlight.Delete(cand.ChannelID + ":" + cand.UserID)
|
||||
|
||||
extractCtx, cancel := context.WithTimeout(ctx, 2*time.Minute)
|
||||
defer cancel()
|
||||
|
||||
if err := s.extractor.Extract(extractCtx, cand.ChannelID, cand.UserID, cand.TeamID, ""); err != nil {
|
||||
log.Printf("⚠ memory extraction failed: channel=%s user=%s: %v",
|
||||
cand.ChannelID, cand.UserID, err)
|
||||
}
|
||||
}(c)
|
||||
}
|
||||
|
||||
wg.Wait()
|
||||
}
|
||||
|
||||
// findCandidates queries for channels with enough new activity since last extraction.
|
||||
func (s *Scanner) findCandidates(ctx context.Context) ([]candidate, error) {
|
||||
db := database.DB
|
||||
if db == nil {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
var query string
|
||||
if database.IsSQLite() {
|
||||
query = `
|
||||
SELECT c.id, c.created_by, COALESCE(cu.team_id, '')
|
||||
FROM channels c
|
||||
LEFT JOIN channel_users cu ON cu.channel_id = c.id AND cu.user_id = c.created_by
|
||||
LEFT JOIN memory_extraction_log mel ON mel.channel_id = c.id AND mel.user_id = c.created_by
|
||||
WHERE c.created_by IS NOT NULL
|
||||
AND (
|
||||
SELECT COUNT(*) FROM messages m
|
||||
WHERE m.channel_id = c.id
|
||||
AND (mel.last_message_id IS NULL OR m.id > mel.last_message_id)
|
||||
) >= 8
|
||||
AND (
|
||||
SELECT MAX(m2.created_at) FROM messages m2 WHERE m2.channel_id = c.id
|
||||
) < datetime('now', '-5 minutes')
|
||||
LIMIT 20
|
||||
`
|
||||
} else {
|
||||
query = `
|
||||
SELECT c.id, c.created_by, COALESCE(cu.team_id, '')
|
||||
FROM channels c
|
||||
LEFT JOIN channel_users cu ON cu.channel_id = c.id AND cu.user_id = c.created_by
|
||||
LEFT JOIN memory_extraction_log mel ON mel.channel_id = c.id AND mel.user_id = c.created_by
|
||||
WHERE c.created_by IS NOT NULL
|
||||
AND (
|
||||
SELECT COUNT(*) FROM messages m
|
||||
WHERE m.channel_id = c.id
|
||||
AND (mel.last_message_id IS NULL OR m.id::text > mel.last_message_id::text)
|
||||
) >= 8
|
||||
AND (
|
||||
SELECT MAX(m2.created_at) FROM messages m2 WHERE m2.channel_id = c.id
|
||||
) < now() - interval '5 minutes'
|
||||
LIMIT 20
|
||||
`
|
||||
}
|
||||
|
||||
rows, err := db.QueryContext(ctx, query)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var candidates []candidate
|
||||
for rows.Next() {
|
||||
var c candidate
|
||||
if err := rows.Scan(&c.ChannelID, &c.UserID, &c.TeamID); err != nil {
|
||||
continue
|
||||
}
|
||||
candidates = append(candidates, c)
|
||||
}
|
||||
|
||||
return candidates, rows.Err()
|
||||
}
|
||||
Reference in New Issue
Block a user