goclaw/internal/http/channel_memory_extraction.go

package http

import (
	"encoding/json"
	"io"
	"net/http"
	"strconv"
	"strings"

	"github.com/google/uuid"
	"github.com/nextlevelbuilder/goclaw/internal/channelmemory"
	"github.com/nextlevelbuilder/goclaw/internal/channels"
	"github.com/nextlevelbuilder/goclaw/internal/i18n"
	"github.com/nextlevelbuilder/goclaw/internal/store"
	"github.com/nextlevelbuilder/goclaw/pkg/protocol"
)

func (h *ChannelInstancesHandler) handleMemoryExtractionStatus(w http.ResponseWriter, r *http.Request) {
	inst := h.memoryInstance(w, r)
	if inst == nil {
		return
	}
	status, err := h.memoryService.Status(r.Context(), inst)
	if err != nil {
		writeError(w, http.StatusInternalServerError, protocol.ErrInternal, "failed to load memory extraction status")
		return
	}
	writeJSON(w, http.StatusOK, status)
}

func (h *ChannelInstancesHandler) handleMemoryExtractionItems(w http.ResponseWriter, r *http.Request) {
	inst := h.memoryInstance(w, r)
	if inst == nil {
		return
	}
	limit := parseMemoryExtractionLimit(r.URL.Query().Get("limit"))
	offset := parseMemoryExtractionOffset(r.URL.Query().Get("offset"))
	opts := store.ChannelMemoryItemListOptions{
		ChannelInstanceID: inst.ID,
		Status:            r.URL.Query().Get("status"),
		Limit:             limit,
		Offset:            offset,
	}
	items, err := h.memoryService.Extractions.ListItems(r.Context(), opts)
	if err != nil {
		writeError(w, http.StatusInternalServerError, protocol.ErrInternal, "failed to list memory extraction items")
		return
	}
	total, err := h.memoryService.Extractions.CountItems(r.Context(), opts)
	if err != nil {
		writeError(w, http.StatusInternalServerError, protocol.ErrInternal, "failed to count memory extraction items")
		return
	}
	writeJSON(w, http.StatusOK, map[string]any{
		"items":  items,
		"total":  total,
		"limit":  limit,
		"offset": offset,
	})
}

func (h *ChannelInstancesHandler) handleMemoryExtractionSettings(w http.ResponseWriter, r *http.Request) {
	if !requireTenantAdmin(w, r, h.tenantStore) {
		return
	}
	inst := h.memoryInstance(w, r)
	if inst == nil {
		return
	}
	body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, 1<<20))
	if err != nil {
		locale := store.LocaleFromContext(r.Context())
		writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidJSON))
		return
	}
	configJSON, normalized, err := channelmemory.ApplyInstanceConfigPatch(inst.Config, body)
	if err != nil {
		locale := store.LocaleFromContext(r.Context())
		writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidJSON))
		return
	}
	if err := h.store.Update(r.Context(), inst.ID, map[string]any{"config": configJSON}); err != nil {
		writeError(w, http.StatusInternalServerError, protocol.ErrInternal, "failed to update memory extraction settings")
		return
	}
	h.emitCacheInvalidate(inst.ID.String())
	emitAudit(h.msgBus, r, "channel_memory.settings_updated", "channel_instance", inst.ID.String())
	writeJSON(w, http.StatusOK, map[string]any{"config": normalized})
}

func (h *ChannelInstancesHandler) handleMemoryExtractionGroups(w http.ResponseWriter, r *http.Request) {
	inst := h.memoryInstance(w, r)
	if inst == nil {
		return
	}
	if inst.ChannelType != channels.TypeDiscord {
		writeJSON(w, http.StatusOK, map[string]any{"groups": []channelmemory.GroupOption{}})
		return
	}
	groups, err := h.memoryService.GroupOptions(r.Context(), inst)
	if err != nil {
		writeError(w, http.StatusInternalServerError, protocol.ErrInternal, "failed to list memory extraction groups")
		return
	}
	writeJSON(w, http.StatusOK, map[string]any{"groups": groups})
}

func (h *ChannelInstancesHandler) handleMemoryExtractionRun(w http.ResponseWriter, r *http.Request) {
	if !requireTenantAdmin(w, r, h.tenantStore) {
		return
	}
	inst := h.memoryInstance(w, r)
	if inst == nil {
		return
	}
	run, err := h.memoryService.RunNow(r.Context(), inst, "manual")
	if err != nil {
		writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, err.Error())
		return
	}
	emitAudit(h.msgBus, r, "channel_memory.run_triggered", "channel_instance", inst.ID.String())
	writeJSON(w, http.StatusAccepted, map[string]any{"run": run})
}

func (h *ChannelInstancesHandler) handleMemoryExtractionRunAll(w http.ResponseWriter, r *http.Request) {
	if !requireTenantAdmin(w, r, h.tenantStore) {
		return
	}
	inst := h.memoryInstance(w, r)
	if inst == nil {
		return
	}
	if acceptsNDJSON(r) {
		h.streamMemoryExtractionRunAll(w, r, inst)
		return
	}
	result, err := h.memoryService.RunAll(r.Context(), inst, "manual_all")
	if err != nil {
		writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, err.Error())
		return
	}
	emitAudit(h.msgBus, r, "channel_memory.run_all_triggered", "channel_instance", inst.ID.String())
	writeJSON(w, http.StatusAccepted, result)
}

func (h *ChannelInstancesHandler) streamMemoryExtractionRunAll(w http.ResponseWriter, r *http.Request, inst *store.ChannelInstanceData) {
	flusher, ok := w.(http.Flusher)
	if !ok {
		writeError(w, http.StatusInternalServerError, protocol.ErrInternal, "streaming is not supported")
		return
	}
	w.Header().Set("Content-Type", "application/x-ndjson")
	w.Header().Set("Cache-Control", "no-cache")
	w.WriteHeader(http.StatusAccepted)
	_, err := h.memoryService.RunAllWithProgress(r.Context(), inst, "manual_all", func(event channelmemory.ProcessAllEvent) error {
		if err := json.NewEncoder(w).Encode(event); err != nil {
			return err
		}
		flusher.Flush()
		return nil
	})
	if err != nil {
		_ = json.NewEncoder(w).Encode(channelmemory.ProcessAllEvent{Type: "error", Error: err.Error()})
		flusher.Flush()
		return
	}
	emitAudit(h.msgBus, r, "channel_memory.run_all_triggered", "channel_instance", inst.ID.String())
}

func acceptsNDJSON(r *http.Request) bool {
	return strings.Contains(r.Header.Get("Accept"), "application/x-ndjson")
}

func parseMemoryExtractionLimit(raw string) int {
	if raw == "" {
		return 20
	}
	n, err := strconv.Atoi(raw)
	if err != nil || n <= 0 {
		return 20
	}
	if n > 200 {
		return 200
	}
	return n
}

func parseMemoryExtractionOffset(raw string) int {
	n, err := strconv.Atoi(raw)
	if err != nil || n < 0 {
		return 0
	}
	return n
}

func (h *ChannelInstancesHandler) handleMemoryExtractionApprove(w http.ResponseWriter, r *http.Request) {
	if !requireTenantAdmin(w, r, h.tenantStore) {
		return
	}
	inst := h.memoryInstance(w, r)
	if inst == nil {
		return
	}
	itemID, ok := parsePathUUID(w, r, "itemID")
	if !ok {
		return
	}
	if !h.memoryItemBelongsToInstance(w, r, itemID, inst.ID) {
		return
	}
	item, err := h.memoryService.Approve(r.Context(), itemID, store.UserIDFromContext(r.Context()))
	if err != nil {
		writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, err.Error())
		return
	}
	emitAudit(h.msgBus, r, "channel_memory.item_approved", "channel_memory_item", itemID.String())
	writeJSON(w, http.StatusOK, item)
}

func (h *ChannelInstancesHandler) handleMemoryExtractionReject(w http.ResponseWriter, r *http.Request) {
	if !requireTenantAdmin(w, r, h.tenantStore) {
		return
	}
	inst := h.memoryInstance(w, r)
	if inst == nil {
		return
	}
	itemID, ok := parsePathUUID(w, r, "itemID")
	if !ok {
		return
	}
	if !h.memoryItemBelongsToInstance(w, r, itemID, inst.ID) {
		return
	}
	if err := h.memoryService.Reject(r.Context(), itemID, store.UserIDFromContext(r.Context())); err != nil {
		writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, err.Error())
		return
	}
	emitAudit(h.msgBus, r, "channel_memory.item_rejected", "channel_memory_item", itemID.String())
	writeJSON(w, http.StatusOK, map[string]string{"status": "rejected"})
}

func (h *ChannelInstancesHandler) handleMemoryExtractionDelete(w http.ResponseWriter, r *http.Request) {
	if !requireTenantAdmin(w, r, h.tenantStore) {
		return
	}
	inst := h.memoryInstance(w, r)
	if inst == nil {
		return
	}
	itemID, ok := parsePathUUID(w, r, "itemID")
	if !ok {
		return
	}
	if !h.memoryItemBelongsToInstance(w, r, itemID, inst.ID) {
		return
	}
	if err := h.memoryService.Delete(r.Context(), itemID); err != nil {
		writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, err.Error())
		return
	}
	emitAudit(h.msgBus, r, "channel_memory.item_deleted", "channel_memory_item", itemID.String())
	writeJSON(w, http.StatusOK, map[string]string{"status": "deleted"})
}

func (h *ChannelInstancesHandler) memoryInstance(w http.ResponseWriter, r *http.Request) *store.ChannelInstanceData {
	id, ok := parsePathUUID(w, r, "id")
	if !ok {
		return nil
	}
	inst, err := h.store.Get(r.Context(), id)
	if err != nil {
		locale := store.LocaleFromContext(r.Context())
		writeError(w, http.StatusNotFound, protocol.ErrNotFound, i18n.T(locale, i18n.MsgInstanceNotFound))
		return nil
	}
	return inst
}

func parsePathUUID(w http.ResponseWriter, r *http.Request, name string) (uuid.UUID, bool) {
	id, err := uuid.Parse(r.PathValue(name))
	if err != nil {
		locale := store.LocaleFromContext(r.Context())
		writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidID, name))
		return uuid.Nil, false
	}
	return id, true
}

func (h *ChannelInstancesHandler) memoryItemBelongsToInstance(w http.ResponseWriter, r *http.Request, itemID, channelID uuid.UUID) bool {
	item, err := h.memoryService.Extractions.GetItem(r.Context(), itemID)
	if err != nil || item.ChannelInstanceID != channelID {
		writeError(w, http.StatusNotFound, protocol.ErrNotFound, "memory extraction item not found")
		return false
	}
	return true
}