package search import ( "bytes" "context" _ "embed" "encoding/json" "fmt" "io" "net/http" "strings" "time" ) // 0001_mail_documents.sql: statisches, versioniertes Schema // (Akzeptanzkriterium 1). Feldnamen hier UND in fields.go müssen // deckungsgleich bleiben — die Konstanten in fields.go sind die einzige // Stelle, aus der Go-Code Feldnamen für Schreib-/Lesezugriffe bezieht. // Manticores SQL-Parser unterstützt keine "--"-Kommentare, daher bleibt // die eingebettete Datei selbst kommentarfrei. // //go:embed migrations/0001_mail_documents.sql var schemaMigration string // SRC-05: Facettenfelder als eigene, nummerierte ALTER-Migrationen // nachgezogen (Manticore erlaubt nur eine Spalte je ALTER TABLE ADD // COLUMN-Anweisung). Reihenfolge ist die Anwendungsreihenfolge. // //go:embed migrations/0002_mail_documents_facets.sql var migrationAddSender string //go:embed migrations/0003_mail_documents_mailbox.sql var migrationAddMailbox string //go:embed migrations/0004_mail_documents_attachment_type.sql var migrationAddAttachmentType string //go:embed migrations/0005_mail_documents_tag.sql var migrationAddTag string // SRC-10: OCR-Sprach-/Qualitätsfelder, gleiches Muster wie die // Facettenfelder aus SRC-05. // //go:embed migrations/0006_mail_documents_ocr_language.sql var migrationAddOCRLanguage string //go:embed migrations/0007_mail_documents_ocr_confidence.sql var migrationAddOCRConfidence string var facetMigrations = []string{ migrationAddSender, migrationAddMailbox, migrationAddAttachmentType, migrationAddTag, migrationAddOCRLanguage, migrationAddOCRConfidence, } // Client spricht ausschließlich über die strukturierte Manticore-HTTP- // JSON-API (kein String-Zusammenbau von SQL-Klauseln, siehe fields.go). // Die SQL-Schnittstelle wird nur für EnsureSchema verwendet, und dort // ausschließlich mit dem statischen, eingebetteten Migrationstext — // niemals mit zur Laufzeit zusammengesetzten Werten. type Client struct { baseURL string http *http.Client } func NewClient(baseURL string) *Client { return &Client{ baseURL: strings.TrimRight(baseURL, "/"), http: &http.Client{Timeout: 10 * time.Second}, } } // EnsureSchema legt den Index gemäß dem versionierten, statischen // Migrationstext an (Akzeptanzkriterium 1) und zieht die Facettenfelder // (SRC-05) idempotent nach. func (c *Client) EnsureSchema(ctx context.Context) error { if err := c.runSchemaSQL(ctx, schemaMigration); err != nil { return fmt.Errorf("search: schema anlegen: %w", err) } for _, migration := range facetMigrations { if err := c.runSchemaSQL(ctx, migration); err != nil { // Manticore meldet bei erneutem ADD COLUMN "field already in // schema" — kein Fehler, sondern der bereits angewendete // Migrationsschritt (Idempotenz, gleiche CREATE-TABLE-IF-NOT- // EXISTS-Konvention wie das Basisschema). if strings.Contains(err.Error(), "already in schema") { continue } return fmt.Errorf("search: facettenfeld-migration: %w", err) } } return nil } func (c *Client) runSchemaSQL(ctx context.Context, query string) error { form := "query=" + query req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.baseURL+"/sql?mode=raw", strings.NewReader(form)) if err != nil { return fmt.Errorf("anfrage bauen: %w", err) } req.Header.Set("Content-Type", "application/x-www-form-urlencoded") resp, err := c.http.Do(req) if err != nil { return fmt.Errorf("ausführen: %w", err) } defer func() { _ = resp.Body.Close() }() body, _ := io.ReadAll(resp.Body) if resp.StatusCode != http.StatusOK { return fmt.Errorf("status %d: %s", resp.StatusCode, string(body)) } if strings.Contains(string(body), `"error":"`) && !strings.Contains(string(body), `"error":""`) { return fmt.Errorf("%s", string(body)) } return nil } // Document ist ein Mail-Suchdokument. Feldnamen im JSON-Tag entsprechen // exakt den Konstanten in fields.go. type Document struct { ID uint64 `json:"-"` TenantSlug string `json:"tenant_slug"` MessageID string `json:"message_id"` Subject string `json:"subject"` Body string `json:"body"` AttachmentText string `json:"attachment_text"` SentAtUnixEpoch int64 `json:"sent_at"` // Facettenfelder (SRC-05), optional — leerer String bedeutet "kein Wert". Sender string `json:"sender"` Mailbox string `json:"mailbox"` AttachmentType string `json:"attachment_type"` Tag string `json:"tag"` // OCR-Sprach-/Qualitätsfelder (SRC-10), optional — leerer String/0 // bedeutet "kein OCR-Anhang bzw. kein Konfidenzwert vorhanden". OCRLanguage string `json:"ocr_language"` OCRConfidence float64 `json:"ocr_confidence"` } // Index legt/ersetzt ein Suchdokument (Akzeptanzkriterium 2: Schreibzugriff // ausschließlich über statische, vordefinierte Feldnamen aus dem // Document-Struct — kein dynamischer Feldname möglich). func (c *Client) Index(ctx context.Context, doc Document) error { payload := map[string]any{ "index": IndexName, "id": doc.ID, "doc": doc, } body, err := json.Marshal(payload) if err != nil { return fmt.Errorf("search: dokument serialisieren: %w", err) } req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.baseURL+"/replace", bytes.NewReader(body)) if err != nil { return fmt.Errorf("search: index-anfrage bauen: %w", err) } req.Header.Set("Content-Type", "application/json") resp, err := c.http.Do(req) if err != nil { return fmt.Errorf("search: dokument indexieren: %w", err) } defer func() { _ = resp.Body.Close() }() respBody, _ := io.ReadAll(resp.Body) if resp.StatusCode != http.StatusOK { return fmt.Errorf("search: dokument indexieren, status %d: %s", resp.StatusCode, string(respBody)) } return nil } // Delete entfernt ein Suchdokument anhand seiner ID (SRC-02 // Akzeptanzkriterium 2: Löschungen werden im Index nachgezogen). Löschen // eines nicht (mehr) vorhandenen Dokuments ist kein Fehler (idempotent). func (c *Client) Delete(ctx context.Context, id uint64) error { payload := map[string]any{ "index": IndexName, "id": id, } body, err := json.Marshal(payload) if err != nil { return fmt.Errorf("search: lösch-anfrage serialisieren: %w", err) } req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.baseURL+"/delete", bytes.NewReader(body)) if err != nil { return fmt.Errorf("search: lösch-anfrage bauen: %w", err) } req.Header.Set("Content-Type", "application/json") resp, err := c.http.Do(req) if err != nil { return fmt.Errorf("search: dokument löschen: %w", err) } defer func() { _ = resp.Body.Close() }() respBody, _ := io.ReadAll(resp.Body) if resp.StatusCode != http.StatusOK { return fmt.Errorf("search: dokument löschen, status %d: %s", resp.StatusCode, string(respBody)) } return nil } // Result ist ein Suchtreffer. type Result struct { MessageID string Subject string Score int64 SentAtUnixEpoch int64 // OCRLanguage/OCRConfidence (SRC-10 Akzeptanzkriterium 1: erkannte // Sprache für Anzeige nutzbar) — leer/0, wenn das Dokument keinen // OCR-Anhang hat. OCRLanguage string OCRConfidence float64 } // fieldWeights gewichtet Betreff höher als Text, Anhangstext am // niedrigsten (SRC-03 Akzeptanzkriterium 1: Ranking berücksichtigt u.a. // Anhangstreffer) — statische Konstanten, keine dynamischen Feldnamen. var fieldWeights = map[string]any{ FieldSubject: 10, FieldBody: 3, FieldAttachmentText: 1, } // doSearchWithSwapRetry führt eine /search-Anfrage aus und wiederholt sie // bis zu zweimal mit kurzer Pause, falls Manticore "unknown local table" // meldet (SRC-09 Akzeptanzkriterium 3: der Reindex-Umschaltmoment // RENAME-alte-Tabelle-weg/RENAME-neue-Tabelle-rein hat ein extrem kurzes // Zeitfenster ohne existierende mail_documents-Tabelle — dieser Retry // überbrückt es, statt eine Suchanfrage in genau diesem Moment fehlschlagen // zu lassen). func (c *Client) doSearchWithSwapRetry(ctx context.Context, body []byte) ([]byte, error) { const maxAttempts = 3 var lastErr error for attempt := 0; attempt < maxAttempts; attempt++ { if attempt > 0 { select { case <-ctx.Done(): return nil, ctx.Err() case <-time.After(20 * time.Millisecond): } } req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.baseURL+"/search", bytes.NewReader(body)) if err != nil { return nil, fmt.Errorf("search: suchanfrage bauen: %w", err) } req.Header.Set("Content-Type", "application/json") resp, err := c.http.Do(req) if err != nil { return nil, fmt.Errorf("search: suche ausführen: %w", err) } respBody, readErr := io.ReadAll(resp.Body) _ = resp.Body.Close() if readErr != nil { return nil, fmt.Errorf("search: antwort lesen: %w", readErr) } if strings.Contains(string(respBody), "unknown local table") { lastErr = fmt.Errorf("search: suche, status %d: %s", resp.StatusCode, string(respBody)) continue } if resp.StatusCode != http.StatusOK { return nil, fmt.Errorf("search: suche, status %d: %s", resp.StatusCode, string(respBody)) } return respBody, nil } return nil, lastErr } // Search sucht queryText innerhalb der Volltextfelder, strikt begrenzt auf // den Mandanten tenantSlug (Akzeptanzkriterium 2: mandantengetrennt // abfragbar) — der Tenant-Filter läuft über ein strukturiertes "equals"- // Feld der JSON-API, niemals über eine interpolierte WHERE-Klausel. // // queryText nutzt Manticores erweiterte Abfragesyntax über den // query_string-Klausel-Typ (Akzeptanzkriterium 3: Phrasensuche mit // Anführungszeichen, Ausschluss mit vorangestelltem "-") — der Wert landet // als reiner JSON-String-Wert, es gibt dabei keinerlei dynamischen // Feld-/Tabellennamen, der beeinflusst werden könnte. Ergebnisse kommen // von Manticore bereits nach Relevanz (BM25, gewichtet über fieldWeights) // absteigend sortiert zurück (Akzeptanzkriterium 1). const searchResultLimit = 1000 func (c *Client) Search(ctx context.Context, tenantSlug, queryText string) ([]Result, error) { payload := map[string]any{ "index": IndexName, "query": map[string]any{ "bool": map[string]any{ "must": []map[string]any{ {"equals": map[string]any{FieldTenantSlug: tenantSlug}}, {"query_string": queryText}, }, }, }, "options": map[string]any{ "field_weights": fieldWeights, }, // Ohne explizites limit begrenzt Manticore standardmäßig auf 20 // Treffer — bei Testkorpora bis 1000 Dokumenten (SRC-03) blieb das // bisher unbemerkt, da nur auf das Vorhandensein einzelner Treffer // geprüft wurde, nicht auf die Gesamtzahl. searchResultLimit deckt // realistische Trefferlisten ab, ohne unbegrenzt zu sein. "limit": searchResultLimit, } body, err := json.Marshal(payload) if err != nil { return nil, fmt.Errorf("search: suchanfrage serialisieren: %w", err) } respBody, err := c.doSearchWithSwapRetry(ctx, body) if err != nil { return nil, err } var parsed searchResponse if err := json.Unmarshal(respBody, &parsed); err != nil { return nil, fmt.Errorf("search: antwort parsen: %w", err) } results := make([]Result, 0, len(parsed.Hits.Hits)) for _, hit := range parsed.Hits.Hits { results = append(results, Result{ MessageID: hit.Source.MessageID, Subject: hit.Source.Subject, Score: hit.Score, SentAtUnixEpoch: hit.Source.SentAtUnixEpoch, OCRLanguage: hit.Source.OCRLanguage, OCRConfidence: hit.Source.OCRConfidence, }) } return results, nil } type searchResponse struct { Hits struct { Hits []struct { Score int64 `json:"_score"` Source struct { MessageID string `json:"message_id"` Subject string `json:"subject"` SentAtUnixEpoch int64 `json:"sent_at"` OCRLanguage string `json:"ocr_language"` OCRConfidence float64 `json:"ocr_confidence"` } `json:"_source"` } `json:"hits"` } `json:"hits"` }