reforger_crawler_main/admin/addons.go

package admin

import (
	"errors"
	"net/http"
	"strings"
	"time"

	"gitea.tbdevent.eu/TBD/reforger_crawler_main/controllers"
	"gitea.tbdevent.eu/TBD/reforger_crawler_main/initializers"
	"gitea.tbdevent.eu/TBD/reforger_crawler_main/metrics"
	"gitea.tbdevent.eu/TBD/reforger_crawler_main/models"
	"github.com/gin-gonic/gin"
	"gorm.io/gorm"
)

// addonRow is the admin view of an addon (no files).
type addonRow struct {
	ID                   string     `json:"id"`
	Name                 string     `json:"name"`
	Type                 string     `json:"type"`
	Author               string     `json:"author"`
	Preview              string     `json:"preview"`
	SubscriberCount      int64      `json:"subscriberCount"`
	CurrentVersionNumber string     `json:"currentVersionNumber"`
	CurrentVersionSize   int64      `json:"currentVersionSize"`
	Unlisted             bool       `json:"unlisted"`
	Private              bool       `json:"private"`
	Blocked              bool       `json:"blocked"`
	ToBeIndexed          bool       `json:"toBeIndexed"`
	IsBeingIndexed       bool       `json:"isBeingIndexed"`
	PriorityIndexing     bool       `json:"priorityIndexing"`
	Recheck              bool       `json:"recheck"`
	IndexStartTime       *time.Time `json:"indexStartTime"`
	LastIndexedAt        *time.Time `json:"lastIndexedAt"`
	CreatedAt            time.Time  `json:"createdAt"`
	UpdatedAt            time.Time  `json:"updatedAt"`
	DeletedAt            *time.Time `json:"deletedAt"`
}

const addonColumns = `id, name, type, author, preview, subscriber_count, current_version_number,
	current_version_size, unlisted, private, blocked, to_be_indexed, is_being_indexed, priority_indexing,
	recheck, index_start_time, last_indexed_at, created_at, updated_at, deleted_at`

var addonSorts = map[string]string{
	"name":        "name",
	"subscribers": "subscriber_count",
	"size":        "current_version_size",
	"created":     "created_at",
	"updated":     "updated_at",
	"indexed":     "last_indexed_at",
	"deleted":     "deleted_at",
}

func listAddons(c *gin.Context) {
	limit := queryInt(c, "limit", 50, 1, 200)
	offset := queryInt(c, "offset", 0, 0, 1<<30)
	status := c.DefaultQuery("status", "all")

	q := initializers.DB.Model(&models.Addon{})
	switch status {
	case "all":
	case "indexed":
		q = q.Where("to_be_indexed = ?", false)
	case "queued":
		q = q.Where("to_be_indexed = ? AND is_being_indexed = ?", true, false)
	case "indexing":
		q = q.Where("is_being_indexed = ?", true)
	case "priority":
		q = q.Where("priority_indexing = ?", true)
	case "recheck":
		q = q.Where("recheck = ?", true)
	case "blocked":
		q = q.Where("blocked = ?", true)
	case "unlisted":
		q = q.Where("unlisted = ?", true)
	case "private":
		q = q.Where("private = ?", true)
	case "deleted":
		q = q.Unscoped().Where("deleted_at IS NOT NULL")
	default:
		fail(c, http.StatusBadRequest, "unknown status")
		return
	}
	if t := c.Query("type"); t != "" {
		q = q.Where("type = ?", t)
	}
	if s := strings.TrimSpace(c.Query("q")); s != "" {
		like := "%" + likeEscape(strings.ToLower(s)) + "%"
		q = q.Where(`id = ? OR LOWER(name) LIKE ? ESCAPE '\' OR LOWER(author) LIKE ? ESCAPE '\'`, s, like, like)
	}

	var total int64
	if err := q.Session(&gorm.Session{}).Count(&total).Error; err != nil {
		fail(c, http.StatusInternalServerError, err.Error())
		return
	}

	sortCol, ok := addonSorts[c.DefaultQuery("sort", "subscribers")]
	if !ok {
		sortCol = "subscriber_count"
	}
	dir := "DESC"
	if c.Query("order") == "asc" {
		dir = "ASC"
	}
	nulls := ""
	if isPostgres() && dir == "DESC" {
		nulls = " NULLS LAST"
	}

	rows := []addonRow{}
	err := q.Select(addonColumns).Order(sortCol + " " + dir + nulls + ", id ASC").Limit(limit).Offset(offset).Scan(&rows).Error
	if err != nil {
		fail(c, http.StatusInternalServerError, err.Error())
		return
	}
	c.JSON(http.StatusOK, gin.H{"items": rows, "total": total, "limit": limit, "offset": offset})
}

type sharedAddon struct {
	AddonID     string `json:"addonId"`
	Name        string `json:"name"`
	SharedFiles int64  `json:"sharedFiles"`
}

func addonDetail(c *gin.Context) {
	id := c.Param("id")
	var row addonRow
	res := initializers.DB.Unscoped().Model(&models.Addon{}).Select(addonColumns).Where("id = ?", id).Limit(1).Scan(&row)
	if res.Error != nil {
		fail(c, http.StatusInternalServerError, res.Error.Error())
		return
	}
	if res.RowsAffected == 0 {
		fail(c, http.StatusNotFound, "addon not found")
		return
	}
	var extra struct {
		Summary          string
		CurrentVersionID int64
	}
	initializers.DB.Unscoped().Model(&models.Addon{}).Select("summary, current_version_id").Where("id = ?", id).Limit(1).Scan(&extra)

	var fileCount, deletedFiles int64
	initializers.DB.Model(&models.AddonFile{}).Where("addon_id = ?", id).Count(&fileCount)
	initializers.DB.Unscoped().Model(&models.AddonFile{}).Where("addon_id = ? AND deleted_at IS NOT NULL", id).Count(&deletedFiles)

	deps, depsTotal, _ := controllers.QueryRelatedAddons(id, "ad.dependency_id", "ad.addon_id", 50, 0)
	dependents, dependentsTotal, _ := controllers.QueryRelatedAddons(id, "ad.addon_id", "ad.dependency_id", 50, 0)

	// declared dependencies that are not (or no longer) in the addons table
	var missingDeps []struct {
		DependencyID   string `json:"id"`
		DependencyName string `json:"name"`
		Version        string `json:"version"`
	}
	initializers.DB.Raw(`SELECT ad.dependency_id, ad.dependency_name, ad.version FROM addon_dependencies ad
		LEFT JOIN addons a ON a.id = ad.dependency_id AND a.deleted_at IS NULL
		WHERE ad.addon_id = ? AND a.id IS NULL LIMIT 50`, id).Scan(&missingDeps)

	shared := []sharedAddon{}
	err := initializers.DB.Raw(`SELECT f.addon_id AS addon_id, COALESCE(MAX(a.name), '') AS name, COUNT(*) AS shared_files
		FROM addon_files f
		LEFT JOIN addons a ON a.id = f.addon_id
		WHERE f.deleted_at IS NULL AND f.addon_id <> ?
		  AND f.hash IN (SELECT s.hash FROM addon_files s WHERE s.addon_id = ? AND s.deleted_at IS NULL)
		  AND NOT EXISTS (SELECT 1 FROM whitelisted_hashes w WHERE w.hash = f.hash AND w.deleted_at IS NULL)
		GROUP BY f.addon_id ORDER BY shared_files DESC LIMIT 25`, id, id).Scan(&shared).Error
	if err != nil {
		fail(c, http.StatusInternalServerError, err.Error())
		return
	}

	c.JSON(http.StatusOK, gin.H{
		"addon":            row,
		"summary":          extra.Summary,
		"currentVersionId": extra.CurrentVersionID,
		"fileCount":        fileCount,
		"deletedFileRows":  deletedFiles,
		"dependencies":     gin.H{"total": depsTotal, "items": deps, "missing": missingDeps},
		"dependents":       gin.H{"total": dependentsTotal, "items": dependents},
		"duplicates":       shared,
	})
}

type fileRow struct {
	ID          uint   `json:"id"`
	Path        string `json:"path"`
	Hash        string `json:"hash"`
	Version     string `json:"version"`
	Whitelisted bool   `json:"whitelisted"`
	OtherAddons int64  `json:"otherAddons"`
}

func addonFiles(c *gin.Context) {
	id := c.Param("id")
	limit := queryInt(c, "limit", 100, 1, 500)
	offset := queryInt(c, "offset", 0, 0, 1<<30)
	q := initializers.DB.Model(&models.AddonFile{}).Where("addon_id = ?", id)
	if s := strings.TrimSpace(c.Query("q")); s != "" {
		like := "%" + likeEscape(strings.ToLower(s)) + "%"
		q = q.Where(`LOWER(path) LIKE ? ESCAPE '\' OR hash = ?`, like, strings.ToLower(s))
	}
	var total int64
	q.Session(&gorm.Session{}).Count(&total)

	rows := []fileRow{}
	if err := q.Select("id, path, hash, version").Order("path ASC").Limit(limit).Offset(offset).Scan(&rows).Error; err != nil {
		fail(c, http.StatusInternalServerError, err.Error())
		return
	}

	if len(rows) > 0 && c.Query("shared") != "0" {
		hashes := make([]string, 0, len(rows))
		for _, r := range rows {
			hashes = append(hashes, r.Hash)
		}
		var wl []string
		initializers.DB.Model(&models.WhitelistedHash{}).Where("hash IN ?", hashes).Pluck("hash", &wl)
		wlSet := map[string]bool{}
		for _, h := range wl {
			wlSet[h] = true
		}
		var counts []struct {
			Hash   string
			Addons int64
		}
		initializers.DB.Raw(`SELECT hash, COUNT(DISTINCT addon_id) AS addons FROM addon_files
			WHERE deleted_at IS NULL AND hash IN ? GROUP BY hash`, hashes).Scan(&counts)
		cnt := map[string]int64{}
		for _, x := range counts {
			cnt[x.Hash] = x.Addons
		}
		for i := range rows {
			rows[i].Whitelisted = wlSet[rows[i].Hash]
			rows[i].OtherAddons = max(0, cnt[rows[i].Hash]-1)
		}
	}
	c.JSON(http.StatusOK, gin.H{"items": rows, "total": total, "limit": limit, "offset": offset})
}

var errNotFound = errors.New("addon not found")

// applyAction changes queue flags of the given addons. UpdateColumns is used
// so updated_at (the queue order) is left alone.
func applyAction(tx *gorm.DB, ids []string, action string, value bool) (int64, error) {
	base := tx.Model(&models.Addon{}).Where("id IN ?", ids)
	var res *gorm.DB
	switch action {
	case "priority":
		cols := map[string]any{"priority_indexing": value}
		if value {
			cols["to_be_indexed"] = true
		}
		res = base.UpdateColumns(cols)
	case "to-be-indexed":
		res = base.UpdateColumns(map[string]any{"to_be_indexed": value})
	case "reindex":
		cols := map[string]any{"to_be_indexed": true, "is_being_indexed": false}
		if value {
			cols["priority_indexing"] = true
		}
		res = base.UpdateColumns(cols)
	case "reset-indexing":
		res = base.UpdateColumns(map[string]any{"is_being_indexed": false})
	case "recheck":
		res = base.UpdateColumns(map[string]any{"recheck": value})
	case "restore":
		res = tx.Unscoped().Model(&models.Addon{}).Where("id IN ? AND deleted_at IS NOT NULL", ids).
			UpdateColumns(map[string]any{"deleted_at": nil, "to_be_indexed": true, "is_being_indexed": false})
	case "delete":
		var addons []models.Addon
		if err := tx.Where("id IN ?", ids).Find(&addons).Error; err != nil {
			return 0, err
		}
		for i := range addons {
			if err := controllers.DeleteAddonCascade(tx, &addons[i]); err != nil {
				return 0, err
			}
		}
		return int64(len(addons)), nil
	default:
		return 0, errors.New("unknown action")
	}
	return res.RowsAffected, res.Error
}

func addonAction(c *gin.Context) {
	id, action := c.Param("id"), c.Param("action")
	var body struct {
		Value *bool `json:"value"`
	}
	_ = c.ShouldBindJSON(&body)
	value := body.Value == nil || *body.Value
	if action == "delete" {
		fail(c, http.StatusBadRequest, "use DELETE /addons/:id")
		return
	}
	n, err := applyAction(initializers.DB, []string{id}, action, value)
	if err != nil {
		fail(c, http.StatusBadRequest, err.Error())
		return
	}
	if n == 0 {
		fail(c, http.StatusNotFound, "addon not found or nothing to change")
		return
	}
	audit(c, "addon_"+strings.ReplaceAll(action, "-", "_"), map[string]any{"addon": id, "value": value})
	invalidate("overview", "queue")
	c.JSON(http.StatusOK, gin.H{"updated": n})
}

func deleteAddon(c *gin.Context) {
	id := c.Param("id")
	err := initializers.DB.Transaction(func(tx *gorm.DB) error {
		var addon models.Addon
		if err := tx.Where("id = ?", id).First(&addon).Error; err != nil {
			return errNotFound
		}
		return controllers.DeleteAddonCascade(tx, &addon)
	})
	if errors.Is(err, errNotFound) {
		fail(c, http.StatusNotFound, err.Error())
		return
	}
	if err != nil {
		fail(c, http.StatusInternalServerError, err.Error())
		return
	}
	audit(c, "addon_delete", map[string]any{"addon": id})
	invalidate("overview", "queue")
	c.JSON(http.StatusOK, gin.H{"deleted": id})
}

// purgeAddon permanently removes an already soft-deleted addon and its rows.
func purgeAddon(c *gin.Context) {
	id := c.Param("id")
	if c.Query("confirm") != id {
		fail(c, http.StatusBadRequest, "pass ?confirm=<addon id> to purge")
		return
	}
	var files int64
	err := initializers.DB.Transaction(func(tx *gorm.DB) error {
		var addon models.Addon
		if err := tx.Unscoped().Where("id = ? AND deleted_at IS NOT NULL", id).First(&addon).Error; err != nil {
			return errNotFound
		}
		res := tx.Unscoped().Where("addon_id = ?", id).Delete(&models.AddonFile{})
		if res.Error != nil {
			return res.Error
		}
		files = res.RowsAffected
		if err := tx.Where("addon_id = ?", id).Delete(&models.AddonDependency{}).Error; err != nil {
			return err
		}
		return tx.Unscoped().Delete(&addon).Error
	})
	if errors.Is(err, errNotFound) {
		fail(c, http.StatusNotFound, "addon not found or not deleted")
		return
	}
	if err != nil {
		fail(c, http.StatusInternalServerError, err.Error())
		return
	}
	audit(c, "addon_purge", map[string]any{"addon": id, "files": files})
	invalidate("overview")
	c.JSON(http.StatusOK, gin.H{"purged": id, "files": files})
}

func bulkAddons(c *gin.Context) {
	var body struct {
		IDs    []string `json:"ids"`
		Action string   `json:"action"`
		Value  *bool    `json:"value"`
	}
	if err := c.ShouldBindJSON(&body); err != nil || len(body.IDs) == 0 {
		fail(c, http.StatusBadRequest, "ids and action are required")
		return
	}
	if len(body.IDs) > 500 {
		fail(c, http.StatusBadRequest, "at most 500 ids per request")
		return
	}
	value := body.Value == nil || *body.Value
	var n int64
	err := initializers.DB.Transaction(func(tx *gorm.DB) error {
		var err error
		n, err = applyAction(tx, body.IDs, body.Action, value)
		return err
	})
	if err != nil {
		fail(c, http.StatusBadRequest, err.Error())
		return
	}
	audit(c, "bulk_"+strings.ReplaceAll(body.Action, "-", "_"), map[string]any{"count": len(body.IDs), "updated": n, "value": value})
	invalidate("overview", "queue")
	c.JSON(http.StatusOK, gin.H{"updated": n})
}

// ---- queue ---------------------------------------------------------------

type queueRow struct {
	ID                 string     `json:"id"`
	Name               string     `json:"name"`
	CurrentVersionSize int64      `json:"currentVersionSize"`
	SubscriberCount    int64      `json:"subscriberCount"`
	PriorityIndexing   bool       `json:"priorityIndexing"`
	Blocked            bool       `json:"blocked"`
	IndexStartTime     *time.Time `json:"indexStartTime"`
	UpdatedAt          time.Time  `json:"updatedAt"`
	ElapsedSec         int64      `json:"elapsedSec"`
	Stuck              bool       `json:"stuck"`
}

const queueColumns = "id, name, current_version_size, subscriber_count, priority_indexing, blocked, index_start_time, updated_at"

func buildQueue() (gin.H, error) {
	counts, err := metrics.LoadQueueCounts()
	if err != nil {
		return nil, err
	}
	var blockedQueued int64
	initializers.DB.Model(&models.Addon{}).Where("to_be_indexed = ? AND blocked = ?", true, true).Count(&blockedQueued)

	now := time.Now()
	indexing := []queueRow{}
	initializers.DB.Model(&models.Addon{}).Select(queueColumns).Where("is_being_indexed = ?", true).
		Order("index_start_time ASC").Limit(100).Scan(&indexing)
	for i := range indexing {
		if t := indexing[i].IndexStartTime; t != nil && !t.IsZero() {
			indexing[i].ElapsedSec = int64(now.Sub(*t).Seconds())
			indexing[i].Stuck = now.Sub(*t) > initializers.IndexTimeout
		}
	}

	priority := []queueRow{}
	initializers.DB.Model(&models.Addon{}).Select(queueColumns).
		Where("to_be_indexed = ? AND is_being_indexed = ? AND priority_indexing = ?", true, false, true).
		Order("updated_at ASC").Limit(50).Scan(&priority)

	// same order the indexers claim jobs in
	next := []queueRow{}
	initializers.DB.Model(&models.Addon{}).Select(queueColumns).
		Where("to_be_indexed = ? AND is_being_indexed = ? AND blocked = ?", true, false, false).
		Order("priority_indexing DESC, updated_at ASC").Limit(25).Scan(&next)

	ts := now.Unix()
	ok := map[string]string{"result": "ok"}
	sel := metrics.Selector{Service: metrics.MainService, Name: "indexer_submissions_total", Match: ok}
	perHour1h := metrics.Sum(sel, ts-3600, ts)
	perHour6h := metrics.Sum(sel, ts-6*3600, ts) / 6
	indexed24h := metrics.Sum(sel, ts-86400, ts)
	var etaHours *float64
	if perHour6h > 0 {
		v := float64(counts.Queued) / perHour6h
		etaHours = &v
	}
	duration, _ := metrics.Totals(metrics.Selector{Service: metrics.MainService, Name: "indexer_job_duration_seconds"}, ts-86400, ts)
	var durationStats any
	if len(duration) > 0 {
		durationStats = duration[0]
	}

	return gin.H{
		"counts":         counts,
		"blockedQueued":  blockedQueued,
		"timeoutMinutes": int(initializers.IndexTimeout.Minutes()),
		"indexing":       indexing,
		"priority":       priority,
		"next":           next,
		"throughput": gin.H{
			"perHour1h":   perHour1h,
			"perHour6h":   perHour6h,
			"indexed24h":  indexed24h,
			"etaHours":    etaHours,
			"duration24h": durationStats,
		},
	}, nil
}

func queue(c *gin.Context) {
	data, at, err := cached("queue", 5*time.Second, buildQueue)
	if err != nil {
		fail(c, http.StatusInternalServerError, err.Error())
		return
	}
	out := gin.H{"asOf": at.Unix()}
	for k, v := range data {
		out[k] = v
	}
	c.JSON(http.StatusOK, out)
}

func resetStuck(c *gin.Context) {
	var body struct {
		OlderThanMinutes int `json:"olderThanMinutes"`
	}
	_ = c.ShouldBindJSON(&body)
	older := initializers.IndexTimeout
	if body.OlderThanMinutes > 0 {
		older = time.Duration(body.OlderThanMinutes) * time.Minute
	}
	n := controllers.ResetTimedOutIndexing(older)
	metrics.IndexerTimeoutsReset.Add(float64(n))
	audit(c, "queue_reset_stuck", map[string]any{"olderThanMinutes": older.Minutes(), "reset": n})
	invalidate("queue", "overview")
	c.JSON(http.StatusOK, gin.H{"reset": n})
}