nexarch/mail/internal/search/facets.go

// SRC-05: Facetten- & Filter-API. Nutzt Manticores strukturierte
// Aggregations-API (aggs.terms/aggs.range) — keine dynamische
// SQL-Klauselbildung, dieselbe Konvention wie Search/Delete (fields.go).
package search

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"time"
)

// FacetFilter schränkt Suche/Facettenberechnung auf einen bereits
// gewählten Facettenwert ein. Field ist der geschlossene FacetField-Typ
// (SRC-11) — buildFilteredMust prüft zusätzlich FacetField.IsValid(),
// sodass selbst ein über json.Unmarshal aus der Datenbank
// rekonstruierter, nicht mehr gültiger Wert (z. B. nach Entfernen eines
// Feldes) abgelehnt wird statt stillschweigend durchzulaufen.
type FacetFilter struct {
	Field FacetField
	Value string
}

// FacetValue ist ein einzelner Facettenwert mit Trefferzahl
// (Akzeptanzkriterium 1).
type FacetValue struct {
	Value string
	Count int64
}

// DateRangeFacet ist ein Zeitraum-Bucket mit Trefferzahl.
type DateRangeFacet struct {
	Label string
	Count int64
}

// FacetResult fasst alle Facettendimensionen einer Anfrage zusammen.
type FacetResult struct {
	// Values ist je FacetFields-Eintrag (sender/mailbox/attachment_type/tag)
	// befüllt.
	Values map[string][]FacetValue
	// DateRanges sind feste Zeitraum-Buckets über FieldSentAt.
	DateRanges []DateRangeFacet
}

// farFuture ist die obere Grenze des jüngsten Zeitraum-Buckets. Manticores
// range-Aggregation verlangt für jeden Bucket ein explizites "to" — ein
// hinreichend großer fester Wert (Jahr 2100) übernimmt die Rolle von
// "unbegrenzt in die Zukunft", ohne den Feldtyp zu wechseln.
const farFuture int64 = 4102444800

type dateRangeBoundary struct {
	label string
	from  int64 // 0 = ab Epoch (unbegrenzt in die Vergangenheit)
	to    int64
}

// dateRangeBoundaries berechnet die festen Zeitraum-Buckets relativ zu now
// (Parameter statt time.Now() direkt, damit Facets testbar bleibt).
func dateRangeBoundaries(now time.Time) []dateRangeBoundary {
	sevenDaysAgo := now.AddDate(0, 0, -7).Unix()
	thirtyDaysAgo := now.AddDate(0, 0, -30).Unix()
	oneYearAgo := now.AddDate(-1, 0, 0).Unix()
	return []dateRangeBoundary{
		{label: "letzte_7_tage", from: sevenDaysAgo, to: farFuture},
		{label: "letzte_30_tage", from: thirtyDaysAgo, to: sevenDaysAgo},
		{label: "letztes_jahr", from: oneYearAgo, to: thirtyDaysAgo},
		{label: "aelter", from: 0, to: oneYearAgo},
	}
}

// buildFilteredMust baut die gemeinsame bool.must-Liste für Facets und
// SearchWithFilters: Tenant-Filter zwingend, optionaler Suchtext, dann je
// Filter eine zusätzliche equals-Klausel (UND-Verknüpfung) — einzige
// Stelle, an der Filter-Feldnamen geprüft werden, über das geschlossene
// FacetField.IsValid() (SRC-11 Akzeptanzkriterium 2), nicht über eine
// durchsuchbare Liste.
func buildFilteredMust(tenantSlug, queryText string, filters []FacetFilter) ([]map[string]any, error) {
	must := []map[string]any{
		{"equals": map[string]any{FieldTenantSlug: tenantSlug}},
	}
	if queryText != "" {
		must = append(must, map[string]any{"query_string": queryText})
	}
	for _, f := range filters {
		if !f.Field.IsValid() {
			return nil, fmt.Errorf("search: unbekanntes facettenfeld %q", f.Field)
		}
		must = append(must, map[string]any{"equals": map[string]any{string(f.Field): f.Value}})
	}
	return must, nil
}

// SearchWithFilters ist Search, zusätzlich beschränkt auf Dokumente, die
// ALLE angegebenen Filter erfüllen (UND-Verknüpfung, dieselbe Semantik wie
// Facets) — Grundlage für SRC-08s Wiederausführung gespeicherter Suchen
// mit Filtern.
func (c *Client) SearchWithFilters(ctx context.Context, tenantSlug, queryText string, filters []FacetFilter) ([]Result, error) {
	if len(filters) == 0 {
		return c.Search(ctx, tenantSlug, queryText)
	}
	must, err := buildFilteredMust(tenantSlug, queryText, filters)
	if err != nil {
		return nil, err
	}

	payload := map[string]any{
		"index": IndexName,
		"query": map[string]any{"bool": map[string]any{"must": must}},
		"options": map[string]any{
			"field_weights": fieldWeights,
		},
		"limit": searchResultLimit,
	}
	body, err := json.Marshal(payload)
	if err != nil {
		return nil, fmt.Errorf("search: gefilterte 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
}

// Facets berechnet für jede Filterdimension (Akzeptanzkriterium 1) die
// Trefferzahl je Wert, mandantengetrennt (Akzeptanzkriterium 3) und unter
// Berücksichtigung bereits gewählter Filter (Akzeptanzkriterium 2: mehrere
// Filter kombinieren sich als UND-Verknüpfung in derselben bool.must-Liste
// wie der Tenant-Filter).
func (c *Client) Facets(ctx context.Context, tenantSlug, queryText string, filters []FacetFilter) (FacetResult, error) {
	must, err := buildFilteredMust(tenantSlug, queryText, filters)
	if err != nil {
		return FacetResult{}, err
	}

	aggs := map[string]any{}
	for _, field := range FacetFields {
		aggs[string(field)] = map[string]any{"terms": map[string]any{"field": string(field), "size": 100}}
	}
	boundaries := dateRangeBoundaries(time.Now())
	ranges := make([]map[string]any, 0, len(boundaries))
	for _, b := range boundaries {
		ranges = append(ranges, map[string]any{"from": b.from, "to": b.to})
	}
	aggs["sent_at"] = map[string]any{"range": map[string]any{"field": FieldSentAt, "ranges": ranges}}

	payload := map[string]any{
		"index": IndexName,
		"query": map[string]any{"bool": map[string]any{"must": must}},
		"aggs":  aggs,
		"limit": 0,
	}
	body, err := json.Marshal(payload)
	if err != nil {
		return FacetResult{}, fmt.Errorf("search: facettenanfrage serialisieren: %w", err)
	}

	req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.baseURL+"/search", bytes.NewReader(body))
	if err != nil {
		return FacetResult{}, fmt.Errorf("search: facettenanfrage bauen: %w", err)
	}
	req.Header.Set("Content-Type", "application/json")

	resp, err := c.http.Do(req)
	if err != nil {
		return FacetResult{}, fmt.Errorf("search: facetten abrufen: %w", err)
	}
	defer func() { _ = resp.Body.Close() }()
	respBody, err := io.ReadAll(resp.Body)
	if err != nil {
		return FacetResult{}, fmt.Errorf("search: facetten-antwort lesen: %w", err)
	}
	if resp.StatusCode != http.StatusOK {
		return FacetResult{}, fmt.Errorf("search: facetten, status %d: %s", resp.StatusCode, string(respBody))
	}

	var parsed facetResponse
	if err := json.Unmarshal(respBody, &parsed); err != nil {
		return FacetResult{}, fmt.Errorf("search: facetten-antwort parsen: %w", err)
	}

	result := FacetResult{Values: make(map[string][]FacetValue, len(FacetFields))}
	for _, field := range FacetFields {
		bucket := parsed.Aggregations[string(field)]
		values := make([]FacetValue, 0, len(bucket.Buckets))
		for _, b := range bucket.Buckets {
			if b.Key == "" {
				continue
			}
			values = append(values, FacetValue{Value: b.Key, Count: b.DocCount})
		}
		result.Values[string(field)] = values
	}

	sentAtBucket := parsed.Aggregations["sent_at"]
	result.DateRanges = make([]DateRangeFacet, 0, len(boundaries))
	for i, b := range boundaries {
		count := int64(0)
		if i < len(sentAtBucket.Buckets) {
			count = sentAtBucket.Buckets[i].DocCount
		}
		result.DateRanges = append(result.DateRanges, DateRangeFacet{Label: b.label, Count: count})
	}

	return result, nil
}

// AttachmentsBelowConfidence liefert alle Dokumente eines Mandanten, deren
// OCR-Konfidenzwert UNTER threshold liegt (SRC-10 Akzeptanzkriterium 3:
// gezielt für manuelle Nachbearbeitung auffindbar). Dokumente ohne
// OCR-Anhang (ocr_confidence bleibt 0) tauchen hier NICHT auf — 0 ist
// "kein Wert", nicht "schlechtester Wert" (siehe Document-Feldkommentar) —
// daher zusätzlicher Filter ocr_confidence > 0.
func (c *Client) AttachmentsBelowConfidence(ctx context.Context, tenantSlug string, threshold float64) ([]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}},
					{"range": map[string]any{FieldOCRConfidence: map[string]any{"gt": 0}}},
					{"range": map[string]any{FieldOCRConfidence: map[string]any{"lt": threshold}}},
				},
			},
		},
		"sort":  []map[string]any{{FieldOCRConfidence: "asc"}},
		"limit": searchResultLimit,
	}
	body, err := json.Marshal(payload)
	if err != nil {
		return nil, fmt.Errorf("search: konfidenzanfrage 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 facetResponse struct {
	Aggregations map[string]struct {
		Buckets []struct {
			Key      string `json:"key"`
			DocCount int64  `json:"doc_count"`
		} `json:"buckets"`
	} `json:"aggregations"`
}