sysopsandClaude Sonnet 5
26c0e04c75
emails_global bekam nie Tenant-Mails gespiegelt, Superadmin-Suche sah nur
~30% aller Mails. Zusätzlich sanken Mails ohne gültigen Date-Header
(date_ts=0) beim datumssortierten Listing ans Ende aller Ergebnisse und
waren dadurch praktisch unauffindbar ("GUI zeigt neue Mails nicht") -
Import/Indexierung liefen technisch korrekt, nur Sortierung/Spiegelung
waren kaputt.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UPFC6Jk2ke1Pq9XcuVGP1R
113 lines
2.9 KiB
Go
113 lines
2.9 KiB
Go
package index
|
|
|
|
import (
|
|
"fmt"
|
|
"log/slog"
|
|
"sync"
|
|
)
|
|
|
|
// TenantIndexWorker processes MailDocument indexing requests asynchronously,
|
|
// routing each document to the correct per-tenant index via TenantIndexer.
|
|
type TenantIndexWorker struct {
|
|
mgr TenantIndexer
|
|
queue chan MailDocument
|
|
done chan struct{}
|
|
wg sync.WaitGroup
|
|
logger *slog.Logger
|
|
}
|
|
|
|
// NewTenantWorker creates a new TenantIndexWorker with the given queue capacity.
|
|
func NewTenantWorker(mgr TenantIndexer, queueSize int, logger *slog.Logger) *TenantIndexWorker {
|
|
if queueSize <= 0 {
|
|
queueSize = 1000
|
|
}
|
|
return &TenantIndexWorker{
|
|
mgr: mgr,
|
|
queue: make(chan MailDocument, queueSize),
|
|
done: make(chan struct{}),
|
|
logger: logger,
|
|
}
|
|
}
|
|
|
|
// Submit enqueues a document for background indexing. If the queue is full,
|
|
// the document is dropped and a warning is logged.
|
|
func (w *TenantIndexWorker) Submit(doc MailDocument) {
|
|
select {
|
|
case w.queue <- doc:
|
|
// queued
|
|
default:
|
|
w.logger.Warn("tenant index worker: queue full, dropping document", "id", doc.ID)
|
|
}
|
|
}
|
|
|
|
// Start launches the background goroutine that processes the queue.
|
|
func (w *TenantIndexWorker) Start() {
|
|
w.wg.Add(1)
|
|
go func() {
|
|
defer w.wg.Done()
|
|
w.logger.Info("tenant index worker: started", "queue_size", cap(w.queue))
|
|
for {
|
|
select {
|
|
case doc, ok := <-w.queue:
|
|
if !ok {
|
|
return
|
|
}
|
|
w.indexDoc(doc)
|
|
case <-w.done:
|
|
// Drain remaining items in the queue before exiting.
|
|
for {
|
|
select {
|
|
case doc, ok := <-w.queue:
|
|
if !ok {
|
|
return
|
|
}
|
|
w.indexDoc(doc)
|
|
default:
|
|
return
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}()
|
|
}
|
|
|
|
// Stop signals the worker to drain remaining items and stop.
|
|
func (w *TenantIndexWorker) Stop() {
|
|
close(w.done)
|
|
w.wg.Wait()
|
|
w.logger.Info("tenant index worker: stopped")
|
|
}
|
|
|
|
// QueueLen returns the current number of items waiting in the queue.
|
|
func (w *TenantIndexWorker) QueueLen() int {
|
|
return len(w.queue)
|
|
}
|
|
|
|
func (w *TenantIndexWorker) indexDoc(doc MailDocument) {
|
|
// A panic while indexing a single document must not kill the long-running
|
|
// worker goroutine (and with it the whole process).
|
|
defer func() {
|
|
if r := recover(); r != nil {
|
|
w.logger.Error("tenant index worker: recovered from panic",
|
|
"id", doc.ID, "tenant_id", doc.TenantID, "panic", fmt.Sprintf("%v", r))
|
|
}
|
|
}()
|
|
|
|
idx := w.mgr.ForTenant(doc.TenantID)
|
|
if idx == nil {
|
|
w.logger.Error("tenant index worker: no indexer for tenant", "id", doc.ID, "tenant_id", doc.TenantID)
|
|
return
|
|
}
|
|
if err := idx.IndexSync(doc); err != nil {
|
|
w.logger.Error("tenant index worker: index failed", "id", doc.ID, "tenant_id", doc.TenantID, "err", err)
|
|
return
|
|
}
|
|
// Superadmin/global search reads emails_global — mirror every
|
|
// tenant-scoped mail there too, not just untenanted ones.
|
|
if doc.TenantID != nil {
|
|
if err := w.mgr.Global().IndexSync(doc); err != nil {
|
|
w.logger.Error("tenant index worker: global index failed", "id", doc.ID, "err", err)
|
|
}
|
|
}
|
|
}
|