server/internal/handlers/agent_push_batch.go

Rama main · réplica de lectura del repositorio original.

package handlers

import (
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"strings"

	"blackbox/server/internal/hub"
	"blackbox/server/internal/middleware"
	"blackbox/server/internal/models"
	"blackbox/shared/types"
	"gorm.io/gorm"
)

const (
	maxBatchSize           = 200
	maxAgentBatchBodyBytes = 10 << 20 // 10 MB
)

type batchPushResponse struct {
	Accepted []string         `json:"accepted"`
	Failed   []batchPushError `json:"failed"`
}

type batchPushError struct {
	ID        string `json:"id"`
	Reason    string `json:"reason"`
	Permanent bool   `json:"permanent"`
}

// AgentPushBatch accepts a JSON array of entries and ingests each through the
// same validation and upsert logic as AgentPush. Returns 200 with accepted/failed
// lists. Only returns 4xx for structural problems (oversized body, batch > 200).
func AgentPushBatch(database *gorm.DB, h *hub.Hub, incidentCh chan<- types.Entry, shutdown <-chan struct{}) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		nodeName, ok := middleware.AgentNodeFromContext(r.Context())
		if !ok {
			writeError(w, http.StatusUnauthorized, "unauthorized")
			return
		}

		var entries []types.Entry
		if !decodeJSONBody(w, r, maxAgentBatchBodyBytes, &entries) {
			return
		}

		if len(entries) > maxBatchSize {
			writeError(w, http.StatusBadRequest, fmt.Sprintf("batch too large: max %d entries", maxBatchSize))
			return
		}

		resp := batchPushResponse{
			Accepted: make([]string, 0, len(entries)),
			Failed:   make([]batchPushError, 0),
		}

		var excludedTargets []models.ExcludedTarget
		if err := database.Find(&excludedTargets).Error; err != nil {
			// fail closed — do not process batch without knowing the exclude list
			writeError(w, http.StatusInternalServerError, "failed to load exclusion list")
			return
		}

		nodeUpdated := false

		for _, entry := range entries {
			if entry.ID == "" {
				resp.Failed = append(resp.Failed, batchPushError{ID: entry.ID, Reason: "entry id is required", Permanent: true})
				continue
			}
			if entry.NodeName != "" && entry.NodeName != nodeName {
				resp.Failed = append(resp.Failed, batchPushError{ID: entry.ID, Reason: "agent node mismatch", Permanent: true})
				continue
			}
			entry.NodeName = nodeName
			serviceName := strings.ToLower(strings.TrimSpace(entry.Service))
			if serviceName == "" && !isAgentMetaEvent(entry) {
				resp.Failed = append(resp.Failed, batchPushError{ID: entry.ID, Reason: "service is required", Permanent: true})
				continue
			}
			entry.Service = serviceName
			if isExcludedInMemory(excludedTargets, entry) {
				resp.Accepted = append(resp.Accepted, entry.ID)
				continue
			}

			if entry.Source == "docker" && entry.Event == "restart" {
				res := replaceDockerRestartEntry(database, entry, nodeName, h, incidentCh, shutdown)
				if res.Err != nil {
					resp.Failed = append(resp.Failed, batchPushError{ID: entry.ID, Reason: "failed to process restart entry"})
					continue
				}
				if res.Found {
					if upsertNode(database, res.Updated) {
						nodeUpdated = true
					}
					resp.Accepted = append(resp.Accepted, entry.ID)
					continue
				}
			}

			created, err := createEntryIdempotent(database, entry, "agent-push-batch")
			if errors.Is(err, errEntryIDConflict) {
				resp.Failed = append(resp.Failed, batchPushError{ID: entry.ID, Reason: "entry id already exists", Permanent: true})
				continue
			}
			if err != nil {
				resp.Failed = append(resp.Failed, batchPushError{ID: entry.ID, Reason: "failed to save entry"})
				continue
			}
			if created {
				dispatchToIncidentChannelWithShutdown(incidentCh, shutdown, entry)
				if h != nil {
					if msg := MarshalWSMessage("entry", entry); msg != nil {
						h.Broadcast(msg)
					}
				}
			}
			if upsertNode(database, entry) {
				nodeUpdated = true
			}
			resp.Accepted = append(resp.Accepted, entry.ID)
		}

		if nodeUpdated {
			broadcastNodeStatus(database, h)
		}

		w.Header().Set("Content-Type", "application/json")
		if err := json.NewEncoder(w).Encode(resp); err != nil {
			return
		}
	}
}