glpi-neuroforge-mega/platform/neuroforge/internal/httpapi/integration.go

package httpapi

import (
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"fmt"
	"net/http"
	"sort"
	"strings"

	"neuroforge/internal/core"
)

type integrationKnowledgeChunk struct {
	Index       int       `json:"index"`
	Text        string    `json:"text"`
	Vector      []float32 `json:"vector"`
	ContentHash string    `json:"content_hash,omitempty"`
}

type integrationKnowledgeUpsert struct {
	Namespace  string                      `json:"namespace"`
	DocumentID string                      `json:"document_id"`
	Title      string                      `json:"title,omitempty"`
	SourceURI  string                      `json:"source_uri,omitempty"`
	Tags       []string                    `json:"tags,omitempty"`
	Confidence float64                     `json:"confidence,omitempty"`
	Chunks     []integrationKnowledgeChunk `json:"chunks"`
}

type integrationKnowledgeSearch struct {
	Namespace     string    `json:"namespace"`
	Vector        []float32 `json:"vector"`
	K             int       `json:"k"`
	MinSimilarity *float64  `json:"min_similarity,omitempty"`
}

type integrationEventRequest struct {
	Type    string `json:"type"`
	Source  string `json:"source"`
	Message string `json:"message,omitempty"`
	Query   string `json:"query,omitempty"`
	Hits    []struct {
		ID    string  `json:"id"`
		Score float64 `json:"score,omitempty"`
	} `json:"hits,omitempty"`
	Metadata map[string]any `json:"metadata,omitempty"`
}

func integrationSource(namespace string) string {
	return "integration:" + strings.ToLower(strings.TrimSpace(namespace))
}

func integrationMemoryID(namespace, documentID string, chunk int) string {
	sum := sha256.Sum256([]byte(strings.ToLower(strings.TrimSpace(namespace)) + "\x00" + strings.TrimSpace(documentID) + fmt.Sprintf("\x00chunk\x00%d", chunk)))
	return "ik_" + hex.EncodeToString(sum[:16])
}

func validIntegrationName(v string) bool {
	v = strings.TrimSpace(v)
	if v == "" || len(v) > 128 {
		return false
	}
	for _, r := range v {
		if !(r >= 'a' && r <= 'z' || r >= 'A' && r <= 'Z' || r >= '0' && r <= '9' || r == '-' || r == '_' || r == '.' || r == ':') {
			return false
		}
	}
	return true
}

func (s *Server) integrationKnowledgeUpsert(w http.ResponseWriter, r *http.Request) {
	var q integrationKnowledgeUpsert
	if err := decode(r, &q); err != nil {
		s.err(w, http.StatusBadRequest, err)
		return
	}
	q.Namespace = strings.TrimSpace(q.Namespace)
	q.DocumentID = strings.TrimSpace(q.DocumentID)
	if !validIntegrationName(q.Namespace) || !validIntegrationName(q.DocumentID) {
		s.err(w, http.StatusBadRequest, errors.New("namespace/document_id contains unsupported characters"))
		return
	}
	if len(q.Chunks) == 0 || len(q.Chunks) > 512 {
		s.err(w, http.StatusBadRequest, errors.New("chunks must contain 1..512 entries"))
		return
	}
	source := integrationSource(q.Namespace)
	confidence := q.Confidence
	if confidence <= 0 {
		confidence = 1
	}
	if confidence > 1 {
		confidence = 1
	}

	// Snapshot once so document replacement is O(total memories + chunks), not
	// O(chunks * total memories), and batch deletes rebuild ANN indexes once.
	existing := make(map[string]core.Memory)
	for _, m := range s.store.MemoriesSnapshot() {
		if m.Provenance.Source == source && m.Provenance.SourceMemoryID == q.DocumentID && m.Kind == "knowledge.chunk" {
			existing[m.ID] = m
		}
	}

	desired := make(map[string]bool, len(q.Chunks))
	createItems := make([]core.Memory, 0, len(q.Chunks))
	deleteIDs := make([]string, 0, len(existing))
	created, updated, unchanged := 0, 0, 0
	seenIndexes := make(map[int]struct{}, len(q.Chunks))
	sort.Slice(q.Chunks, func(i, j int) bool { return q.Chunks[i].Index < q.Chunks[j].Index })
	for _, chunk := range q.Chunks {
		if chunk.Index < 0 || strings.TrimSpace(chunk.Text) == "" || len(chunk.Vector) == 0 {
			s.err(w, http.StatusBadRequest, errors.New("each chunk requires non-negative index, text and vector"))
			return
		}
		if _, duplicate := seenIndexes[chunk.Index]; duplicate {
			s.err(w, http.StatusBadRequest, fmt.Errorf("duplicate chunk index %d", chunk.Index))
			return
		}
		seenIndexes[chunk.Index] = struct{}{}
		id := integrationMemoryID(q.Namespace, q.DocumentID, chunk.Index)
		desired[id] = true
		current, exists := existing[id]
		if exists && current.Provenance.ContentHash == chunk.ContentHash && current.Provenance.SourceMemoryID == q.DocumentID && len(current.Vector) == len(chunk.Vector) {
			unchanged++
			continue
		}
		if exists {
			deleteIDs = append(deleteIDs, id)
			updated++
		} else {
			created++
		}
		tags := append([]string(nil), q.Tags...)
		tags = append(tags, "integration", "namespace:"+q.Namespace, "document:"+q.DocumentID, "record:chunk")
		createItems = append(createItems, core.Memory{
			ID: id, Kind: "knowledge.chunk", MemoryType: core.MemorySemantic,
			Text: chunk.Text, Vector: append([]float32(nil), chunk.Vector...), Tags: tags,
			Salience: 1, Confidence: confidence,
			Provenance: core.MemoryProvenance{
				Source: source, Actor: "knowledge-sync", SourceMemoryID: q.DocumentID,
				SourceURI: strings.TrimSpace(q.SourceURI), SourceTitle: strings.TrimSpace(q.Title),
				ChunkIndex: chunk.Index, ChunkCount: len(q.Chunks), ContentHash: chunk.ContentHash,
			},
		})
	}
	deleted := 0
	for id := range existing {
		if !desired[id] {
			deleteIDs = append(deleteIDs, id)
			deleted++
		}
	}
	if err := s.store.DeleteMemoriesBatch(deleteIDs); err != nil {
		s.err(w, http.StatusInternalServerError, err)
		return
	}
	if err := s.store.AddMemoriesBatch(createItems); err != nil {
		s.err(w, http.StatusInternalServerError, err)
		return
	}
	_ = s.store.AddKnowledgeEvent(core.KnowledgeEvent{
		Type: "integration.knowledge.synced", Summary: "External knowledge document synchronized", Actor: q.Namespace,
		Metadata: map[string]string{"namespace": q.Namespace, "document_id": q.DocumentID, "created": fmt.Sprint(created), "updated": fmt.Sprint(updated), "deleted": fmt.Sprint(deleted), "unchanged": fmt.Sprint(unchanged)},
	})
	s.json(w, http.StatusOK, map[string]any{"ok": true, "document_id": q.DocumentID, "created": created, "updated": updated, "deleted": deleted, "unchanged": unchanged})
}

func (s *Server) integrationKnowledgeDelete(w http.ResponseWriter, r *http.Request) {
	namespace := strings.TrimSpace(r.PathValue("namespace"))
	documentID := strings.TrimSpace(r.PathValue("document_id"))
	if !validIntegrationName(namespace) || !validIntegrationName(documentID) {
		s.err(w, http.StatusBadRequest, errors.New("invalid namespace or document id"))
		return
	}
	source := integrationSource(namespace)
	ids := make([]string, 0)
	for _, m := range s.store.MemoriesSnapshot() {
		if m.Provenance.Source == source && m.Provenance.SourceMemoryID == documentID && m.Kind == "knowledge.chunk" {
			ids = append(ids, m.ID)
		}
	}
	if err := s.store.DeleteMemoriesBatch(ids); err != nil {
		s.err(w, http.StatusInternalServerError, err)
		return
	}
	deleted := len(ids)
	_ = s.store.AddKnowledgeEvent(core.KnowledgeEvent{Type: "integration.knowledge.deleted", Summary: "External knowledge document removed", Actor: namespace, Metadata: map[string]string{"namespace": namespace, "document_id": documentID, "deleted": fmt.Sprint(deleted)}})
	s.json(w, http.StatusOK, map[string]any{"ok": true, "deleted": deleted})
}

func (s *Server) integrationKnowledgeSearch(w http.ResponseWriter, r *http.Request) {
	var q integrationKnowledgeSearch
	if err := decode(r, &q); err != nil {
		s.err(w, http.StatusBadRequest, err)
		return
	}
	q.Namespace = strings.TrimSpace(q.Namespace)
	if !validIntegrationName(q.Namespace) || len(q.Vector) == 0 {
		s.err(w, http.StatusBadRequest, errors.New("namespace and vector are required"))
		return
	}
	if q.K <= 0 {
		q.K = 128
	}
	if q.K > 500 {
		q.K = 500
	}
	min := -1.0
	if q.MinSimilarity != nil {
		min = *q.MinSimilarity
	}
	hits := s.store.SearchVectorByProvenanceSource(q.Vector, q.K, min, 0, integrationSource(q.Namespace))
	s.json(w, http.StatusOK, hits)
}

func (s *Server) integrationEvent(w http.ResponseWriter, r *http.Request) {
	var q integrationEventRequest
	if err := decode(r, &q); err != nil {
		s.err(w, http.StatusBadRequest, err)
		return
	}
	q.Type = strings.TrimSpace(q.Type)
	q.Source = strings.TrimSpace(q.Source)
	if q.Type == "" || q.Source == "" {
		s.err(w, http.StatusBadRequest, errors.New("type and source are required"))
		return
	}
	meta := map[string]string{}
	for k, v := range q.Metadata {
		if strings.TrimSpace(k) != "" {
			meta[k] = fmt.Sprint(v)
		}
	}
	if strings.TrimSpace(q.Query) != "" {
		meta["query"] = q.Query
	}
	if len(q.Hits) > 0 {
		meta["hit_count"] = fmt.Sprint(len(q.Hits))
		limit := len(q.Hits)
		if limit > 8 {
			limit = 8
		}
		for i := 0; i < limit; i++ {
			meta[fmt.Sprintf("hit_%d", i+1)] = fmt.Sprintf("%s:%.4f", q.Hits[i].ID, q.Hits[i].Score)
		}
	}
	if err := s.store.AddKnowledgeEvent(core.KnowledgeEvent{Type: q.Type, Summary: strings.TrimSpace(q.Message), Actor: q.Source, Reason: "integration event", Metadata: meta}); err != nil {
		s.err(w, http.StatusInternalServerError, err)
		return
	}
	s.json(w, http.StatusAccepted, map[string]bool{"ok": true})
}