nexarch/ai/cmd/ai-app/main.go

// Command ai-app buendelt alle bestehenden AI-HTTP-API-Pakete
// (summaryapi, semanticsearchapi, extractapi, residencyapi, routingsuggestapi)
// zu einem einzigen startbaren Dienst (ENG-08), analog zu connect/cmd/connect-app.
// Zusaetzlich startet main den Inferenz-Queue-Worker (ENG-05) mit
// Quota-Enforcement/Verbrauchszaehlung (GOV-04) im Hintergrund.
package main

import (
	"context"
	"errors"
	"log"
	"net/http"
	"os"
	"os/signal"
	"strings"
	"syscall"
	"time"

	"github.com/jackc/pgx/v5/pgxpool"

	"gitea.perlbach24.de/scripte/nexarch/ai/internal/docchat"
	"gitea.perlbach24.de/scripte/nexarch/ai/internal/docchatapi"
	"gitea.perlbach24.de/scripte/nexarch/ai/internal/extract"
	"gitea.perlbach24.de/scripte/nexarch/ai/internal/extractapi"
	"gitea.perlbach24.de/scripte/nexarch/ai/internal/extractconfirmapi"
	"gitea.perlbach24.de/scripte/nexarch/ai/internal/extractmapping"
	"gitea.perlbach24.de/scripte/nexarch/ai/internal/extractvalidate"
	"gitea.perlbach24.de/scripte/nexarch/ai/internal/invoiceextract"
	"gitea.perlbach24.de/scripte/nexarch/ai/internal/metadatapushclient"
	"gitea.perlbach24.de/scripte/nexarch/ai/internal/moduletype"
	"gitea.perlbach24.de/scripte/nexarch/ai/internal/moduletypeapi"
	"gitea.perlbach24.de/scripte/nexarch/ai/internal/provider"
	"gitea.perlbach24.de/scripte/nexarch/ai/internal/provider/ollama"
	"gitea.perlbach24.de/scripte/nexarch/ai/internal/queue"
	"gitea.perlbach24.de/scripte/nexarch/ai/internal/quota"
	"gitea.perlbach24.de/scripte/nexarch/ai/internal/residency"
	"gitea.perlbach24.de/scripte/nexarch/ai/internal/residencyapi"
	"gitea.perlbach24.de/scripte/nexarch/ai/internal/routingfeedback"
	"gitea.perlbach24.de/scripte/nexarch/ai/internal/routingsuggestapi"
	"gitea.perlbach24.de/scripte/nexarch/ai/internal/semanticsearch"
	"gitea.perlbach24.de/scripte/nexarch/ai/internal/semanticsearchapi"
	"gitea.perlbach24.de/scripte/nexarch/ai/internal/summarize"
	"gitea.perlbach24.de/scripte/nexarch/ai/internal/summaryapi"
	"gitea.perlbach24.de/scripte/nexarch/ai/internal/threadsummarize"
	"gitea.perlbach24.de/scripte/nexarch/ai/internal/vectorindex"
	"gitea.perlbach24.de/scripte/nexarch/common/policyclient"
	"gitea.perlbach24.de/scripte/nexarch/common/tenantregistryclient"
)

const workerPollInterval = 2 * time.Second

const defaultListenAddr = "127.0.0.1:9600"
const defaultModel = "default"

type config struct {
	dsn                string
	serviceToken       string
	addr               string
	model              string
	moduleURLs         map[string]string
	coreBaseURL        string
	coreServiceToken   string
	policyURL          string
	policyServiceToken string
	policyDisabled     bool
}

// policyClient liefert nil nur mit NEXARCH_AI_APP_POLICY_DISABLED=1.
func (c config) policyClient() *policyclient.Client {
	if c.policyURL == "" {
		return nil
	}
	return policyclient.New(c.policyURL, c.policyServiceToken)
}

// parseModuleBaseURLs liest "modul=url,modul2=url2" (EXT-05: je Modul eine
// Basis-URL fuer den Rueckschreib-Push in dessen Metadaten-API).
func parseModuleBaseURLs(raw string) map[string]string {
	urls := map[string]string{}
	for _, pair := range strings.Split(raw, ",") {
		pair = strings.TrimSpace(pair)
		if pair == "" {
			continue
		}
		name, url, ok := strings.Cut(pair, "=")
		if !ok || strings.TrimSpace(name) == "" || strings.TrimSpace(url) == "" {
			continue
		}
		urls[strings.TrimSpace(name)] = strings.TrimSpace(url)
	}
	return urls
}

// loadConfig liest die Pflicht- und optionalen Umgebungsvariablen (AK3, AK4):
// eine fehlende Pflichtvariable liefert einen Fehler statt eines stillen
// Programmabbruchs weiter unten im Code.
func loadConfig(getenv func(string) string) (config, error) {
	cfg := config{
		dsn:                getenv("NEXARCH_AI_APP_TENANT_DSN"),
		serviceToken:       getenv("NEXARCH_AI_APP_SERVICE_TOKEN"),
		addr:               getenv("NEXARCH_AI_APP_LISTEN_ADDR"),
		model:              getenv("NEXARCH_AI_APP_MODEL"),
		moduleURLs:         parseModuleBaseURLs(getenv("NEXARCH_AI_APP_MODULE_BASE_URLS")),
		coreBaseURL:        getenv("NEXARCH_AI_APP_CORE_BASE_URL"),
		coreServiceToken:   getenv("NEXARCH_AI_APP_CORE_SERVICE_TOKEN"),
		policyURL:          getenv("NEXARCH_AI_APP_POLICY_URL"),
		policyServiceToken: getenv("NEXARCH_AI_APP_POLICY_SERVICE_TOKEN"),
		policyDisabled:     getenv("NEXARCH_AI_APP_POLICY_DISABLED") == "1",
	}
	if cfg.policyURL != "" && cfg.policyServiceToken == "" {
		return config{}, errors.New("ai-app: NEXARCH_AI_APP_POLICY_SERVICE_TOKEN ist erforderlich, wenn NEXARCH_AI_APP_POLICY_URL gesetzt ist")
	}
	if cfg.dsn == "" {
		return config{}, errors.New("ai-app: NEXARCH_AI_APP_TENANT_DSN ist erforderlich")
	}
	if cfg.serviceToken == "" {
		return config{}, errors.New("ai-app: NEXARCH_AI_APP_SERVICE_TOKEN ist erforderlich")
	}
	if cfg.policyURL == "" && !cfg.policyDisabled {
		return config{}, errors.New("ai-app: NEXARCH_AI_APP_POLICY_URL ist erforderlich (oder NEXARCH_AI_APP_POLICY_DISABLED=1 als Uebergang)")
	}
	if cfg.addr == "" {
		cfg.addr = defaultListenAddr
	}
	if cfg.model == "" {
		cfg.model = defaultModel
	}
	return cfg, nil
}

// wiring haelt alles zusammen, was ein Test wie main() auch ohne echten
// Netzwerk-Listener aufbauen und pruefen koennen muss.
type wiring struct {
	mux        *http.ServeMux
	queue      *queue.Queue
	worker     *queue.Worker
	quotaStore *quota.Store
}

// build verdrahtet alle AI-API-Pakete (AK1) sowie Queue-Worker und
// Quota-Enforcement (AK2) auf einem gemeinsamen Pool/Provider.
func build(pool *pgxpool.Pool, cfg config, prov provider.Provider, workerID string) wiring {
	mux := http.NewServeMux()
	mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) })

	summaryapi.NewServer(
		&summarize.Summarizer{Provider: prov, Model: cfg.model},
		&threadsummarize.Summarizer{Provider: prov, Model: cfg.model},
		cfg.serviceToken,
	).RegisterRoutes(mux)

	searchSvc := semanticsearch.New(prov, vectorindex.New(pool), cfg.model)
	policy := cfg.policyClient()
	if policy == nil {
		log.Print("ai-app: WARNUNG: Rechtepruefung ai.search.read deaktiviert (NEXARCH_AI_APP_POLICY_DISABLED=1)")
	}
	semanticsearchapi.NewServer(searchSvc, cfg.serviceToken).WithPolicy(policy).RegisterRoutes(mux)

	chunkStore := vectorindex.New(pool)
	docchatapi.NewServer(
		docchat.New(searchSvc, prov, cfg.model, chunkStore),
		cfg.serviceToken,
	).RegisterRoutes(mux)

	extractapi.NewServer(
		&invoiceextract.Extractor{Base: &extract.Extractor{Provider: prov, Model: cfg.model}, Provider: prov, Model: cfg.model},
		extractvalidate.RuleSet{},
		cfg.serviceToken,
	).RegisterRoutes(mux)

	residencyapi.NewServer(residency.NewStore(pool), cfg.serviceToken).RegisterRoutes(mux)

	routingsuggestapi.NewServer(cfg.serviceToken).WithFeedback(&routingfeedback.Store{Pool: pool}).RegisterRoutes(mux)

	moduleClients := make(map[string]*metadatapushclient.Client, len(cfg.moduleURLs))
	for name, baseURL := range cfg.moduleURLs {
		moduleClients[name] = metadatapushclient.New(baseURL, cfg.serviceToken)
	}
	extractconfirmapi.NewServer(
		&extractmapping.Registry{Pool: pool},
		moduleClients,
		cfg.serviceToken,
	).RegisterRoutes(mux)

	// GOV-01/INT-12: Objekttyp-Registrierung durch DMS/Mail nur zulaessig,
	// solange Core TEN-01 fuer die Tenant-Verifikation erreichbar konfiguriert
	// ist (GOV-07 fail-closed) - ohne Core-Basis-URL bleibt der Endpunkt
	// unverdrahtet statt eine Registrierung ohne Verifikation zu erlauben.
	if cfg.coreBaseURL != "" {
		verifier := tenantregistryclient.New(cfg.coreBaseURL, cfg.coreServiceToken)
		moduletypeapi.NewServer(&moduletype.Registry{Pool: pool, Verifier: verifier}, cfg.serviceToken).RegisterRoutes(mux)
	} else {
		log.Print("ai-app: NEXARCH_AI_APP_CORE_BASE_URL nicht gesetzt, Objekttyp-Registrierung (GOV-01) deaktiviert")
	}

	quotaStore := quota.NewStore(pool, nil)
	q := queue.NewQueue(pool, 5*time.Minute)
	q.SetQuotaEnforcer(quotaStore)
	worker := queue.NewWorker(q, prov, workerID)
	worker.SetUsageRecorder(quotaStore)

	return wiring{mux: mux, queue: q, worker: worker, quotaStore: quotaStore}
}

func main() {
	cfg, err := loadConfig(os.Getenv)
	if err != nil {
		log.Fatal(err)
	}

	ctx, stopWorker := context.WithCancel(context.Background())
	defer stopWorker()

	pool, err := pgxpool.New(ctx, cfg.dsn)
	if err != nil {
		log.Fatalf("ai-app: datenbankverbindung: %v", err)
	}
	defer pool.Close()

	prov := ollama.NewDefault()
	w := build(pool, cfg, prov, cfg.addr)
	go w.worker.Run(ctx, workerPollInterval)

	srv := &http.Server{
		Addr:              cfg.addr,
		Handler:           w.mux,
		ReadHeaderTimeout: 5 * time.Second,
		ReadTimeout:       15 * time.Second,
		WriteTimeout:      15 * time.Second,
		IdleTimeout:       60 * time.Second,
	}

	go func() {
		log.Printf("ai-app: listening on %s", cfg.addr)
		if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
			log.Fatalf("ai-app: %v", err)
		}
	}()

	stop := make(chan os.Signal, 1)
	signal.Notify(stop, syscall.SIGINT, syscall.SIGTERM)
	<-stop

	stopWorker()
	shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	_ = srv.Shutdown(shutdownCtx)
}