api/controller/cleanup_upload_inventory.go

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

package controller

import (
	"context"
	"net/http"
	"regexp"
	"time"

	"github.com/goodrain/rainbond/db/model"
	guard "github.com/goodrain/rainbond/pkg/cleanup"
	"github.com/goodrain/rainbond/pkg/component/storage"
	httputil "github.com/goodrain/rainbond/util/http"
)

var uploadInventoryEventID = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9_-]{0,63}$`)

type uploadInventoryItem struct {
	StateFingerprint string    `json:"state_fingerprint"`
	ID               string    `json:"id"`
	EventID          string    `json:"event_id"`
	FileName         string    `json:"file_name"`
	Status           string    `json:"status"`
	CreatedAt        time.Time `json:"created_at"`
	UpdatedAt        time.Time `json:"updated_at"`
	ExpiresAt        time.Time `json:"expires_at"`
	Bytes            *int64    `json:"bytes"`
	Objects          *int      `json:"objects"`
	SizeStatus       string    `json:"size_status"`
}

func systemMeasureUploadChunks(ctx context.Context, id string) (storage.UploadChunkUsage, error) {
	configured := storage.Default()
	if configured == nil {
		return storage.UploadChunkUsage{}, guard.ErrCoordinationUnavailable
	}
	measurer, ok := configured.StorageCli.(interface {
		MeasureUploadChunks(context.Context, string) (storage.UploadChunkUsage, error)
	})
	if !ok {
		return storage.UploadChunkUsage{}, guard.ErrCoordinationUnavailable
	}
	return measurer.MeasureUploadChunks(ctx, id)
}

type uploadPackageInventoryItem struct {
	Referenced         *bool  `json:"referenced"`
	ReferencesComplete bool   `json:"references_complete"`
	EventID            string `json:"event_id"`
	Bytes              *int64 `json:"bytes"`
	Objects            *int   `json:"objects"`
	SizeStatus         string `json:"size_status"`
}

func systemMeasureUploadEvent(ctx context.Context, id string) (storage.UploadChunkUsage, error) {
	configured := storage.Default()
	if configured == nil {
		return storage.UploadChunkUsage{}, guard.ErrCoordinationUnavailable
	}
	measurer, ok := configured.StorageCli.(interface {
		MeasureUploadEvent(context.Context, string) (storage.UploadChunkUsage, error)
	})
	if !ok {
		return storage.UploadChunkUsage{}, guard.ErrCoordinationUnavailable
	}
	return measurer.MeasureUploadEvent(ctx, id)
}

// UploadInventory observes chunks of explicitly requested upload events. It is
// internal to the trusted Console and requires CleanupIdentity on its route.
// Console must derive event IDs from its own enterprise/team records. Neither
// expiration nor this measurement proves that an upload is safe to delete.
func (h *CleanupCoordinationHandler) UploadInventory(w http.ResponseWriter, r *http.Request) {
	var body struct {
		EventIDs []string `json:"event_ids"`
	}
	if !coordinationDecode(w, r, &body) {
		return
	}
	seen := map[string]bool{}
	if len(body.EventIDs) == 0 || len(body.EventIDs) > 50 {
		httputil.ReturnError(r, w, 400, "INVALID_UPLOAD_SCOPE")
		return
	}
	for _, id := range body.EventIDs {
		if !uploadInventoryEventID.MatchString(id) || seen[id] {
			httputil.ReturnError(r, w, 400, "INVALID_UPLOAD_SCOPE")
			return
		}
		seen[id] = true
	}
	if h.database == nil || h.measureUploadChunks == nil {
		coordinationError(w, r, guard.ErrCoordinationUnavailable)
		return
	}
	database := h.database()
	if database == nil {
		coordinationError(w, r, guard.ErrCoordinationUnavailable)
		return
	}
	rows := []model.UploadSession{}
	if err := database.Select("id, event_id, file_name, status, uploaded_chunks, storage_path, created_at, updated_at, expires_at").Where("event_id IN (?)", body.EventIDs).Order("id ASC").Limit(501).Find(&rows).Error; err != nil {
		coordinationError(w, r, guard.ErrCoordinationUnavailable)
		return
	}
	if len(rows) > 500 {
		httputil.ReturnError(r, w, 413, "UPLOAD_INVENTORY_LIMIT")
		return
	}
	ctx, cancel := context.WithTimeout(r.Context(), 30*time.Second)
	defer cancel()
	items := make([]uploadInventoryItem, 0, len(rows))
	for _, row := range rows {
		item := uploadInventoryItem{StateFingerprint: guard.UploadSessionFingerprint(row), ID: row.ID, EventID: row.EventID, FileName: row.FileName, Status: row.Status, CreatedAt: row.CreatedAt, UpdatedAt: row.UpdatedAt, ExpiresAt: row.ExpiresAt, SizeStatus: "unavailable"}
		if ctx.Err() == nil {
			usage, err := h.measureUploadChunks(ctx, row.ID)
			if err == nil && usage.Bytes >= 0 && usage.Objects >= 0 {
				item.Bytes = &usage.Bytes
				item.Objects = &usage.Objects
				item.SizeStatus = "measured"
			}
		}
		items = append(items, item)
	}
	packages := make([]uploadPackageInventoryItem, 0, len(body.EventIDs))
	var references map[string]bool
	if h.uploadPackageReferences != nil {
		references, _ = h.uploadPackageReferences(database, body.EventIDs)
	}

	// Measure an event once, even if it has multiple upload/retry sessions.
	// Legacy non-chunk uploads may have no UploadSession row at all.
	for _, eventID := range body.EventIDs {
		item := uploadPackageInventoryItem{EventID: eventID, ReferencesComplete: references != nil, SizeStatus: "unavailable"}
		if used, ok := references[eventID]; ok {
			item.Referenced = &used
		}
		if h.measureUploadEvent != nil && ctx.Err() == nil {
			usage, err := h.measureUploadEvent(ctx, eventID)
			if err == nil && usage.Bytes >= 0 && usage.Objects >= 0 {
				item.Bytes = &usage.Bytes
				item.Objects = &usage.Objects
				item.SizeStatus = "measured"
			}
		}
		packages = append(packages, item)
	}

	binding := ""
	if h.uploadStorageBinding != nil {
		binding, _ = h.uploadStorageBinding()
	}
	writersReady := false
	if h.inspectReferenceWriters != nil {
		writers, err := h.inspectReferenceWriters(ctx)
		if err == nil {
			writersReady, _ = guard.UploadWriterCoverageRegistered(database, writers)
		}
	}
	httputil.ReturnSuccess(r, w, struct {
		StorageFingerprint string                       `json:"storage_fingerprint"`
		WritersReady       bool                         `json:"writers_ready"`
		Packages           []uploadPackageInventoryItem `json:"packages"`
		Protocol           int                          `json:"protocol"`
		Scope              string                       `json:"scope"`
		Items              []uploadInventoryItem        `json:"items"`
	}{binding, writersReady, packages, 1, "upload_chunks", items})
}