feat: enhance job processing with detailed logging and UI improvements. Introduce revision log entries in the Worker to capture job metadata, including source and community details. Update the Svelte UI to display job logs with expandable details, improving user experience and job tracking capabilities.

internal/jobs/worker.go

@@ -2,8 +2,11 @@ package jobs
 
 import (
 	"context"
+	"fmt"
 	"net/http"
+	"net/netip"
 	"os"
+	"sort"
 	"strings"
 	"time"
 
@@ -37,20 +40,29 @@ func mergeBirdPostApplyMeta(j *Job) {
 }
 
 const (
-	KindModuleRefresh   = "module_refresh"
-	KindDeployApply     = "deploy_apply"
+	KindModuleRefresh    = "module_refresh"
+	KindDeployApply      = "deploy_apply"
 	KindRevisionRollback = "revision_rollback"
-	KindBirdReload      = "bird_reload"
+	KindBirdReload       = "bird_reload"
 )
 
 // Worker executes queued jobs against store.Backend (memory or SQL).
 type Worker struct {
-	Store       store.Backend
-	HTTPClient  *http.Client // optional; CDN refresh uses this (default 45s timeout).
+	Store      store.Backend
+	HTTPClient *http.Client // optional; CDN refresh uses this (default 45s timeout).
 	// Registry is set after BootstrapWorkers creates the job queue; used to chain deploy_apply after refresh/rollback.
 	Registry *Registry
 }
 
+type revisionLogEntry struct {
+	Kind        string   `json:"kind"`
+	Source      string   `json:"source"`
+	Community   string   `json:"community"`
+	PrefixCount int      `json:"prefix_count"`
+	Sample      []string `json:"sample,omitempty"`
+	Message     string   `json:"message"`
+}
+
 var defaultWorkerHTTP = &http.Client{Timeout: 45 * time.Second}
 
 func (w *Worker) httpClient() *http.Client {
@@ -90,6 +102,15 @@ func (w *Worker) Process(j *Job) {
 			return
 		}
 		j.mergeMeta(map[string]any{"revision_id": rev})
+		if entries, total, err := w.buildRevisionLogEntries(j.TenantID, rev); err == nil {
+			j.mergeMeta(map[string]any{
+				"log_entries":   entries,
+				"log_total":     total,
+				"log_generated": time.Now().UTC().Format(time.RFC3339Nano),
+			})
+		} else {
+			j.mergeMeta(map[string]any{"log_build_error": err.Error()})
+		}
 		w.enqueueDeployAllSpeakers(j, j.TenantID, rev)
 		j.Succeed()
 	case KindDeployApply:
@@ -169,6 +190,7 @@ func (w *Worker) runDeployApply(j *Job) {
 			return
 		}
 	}
+	applied := make([]string, 0, 8)
 	applyOne := func(speakerID string) error {
 		if err := w.Store.SetLastAppliedRevision(j.TenantID, speakerID, revID); err != nil {
 			return err
@@ -177,6 +199,7 @@ func (w *Worker) runDeployApply(j *Job) {
 		if err := w.Store.PublishRevisionForSpeaker(speakerID, revID); err != nil {
 			return err
 		}
+		applied = append(applied, speakerID)
 		return nil
 	}
 	if hasSpeaker && spk != "" {
@@ -194,6 +217,14 @@ func (w *Worker) runDeployApply(j *Job) {
 			return
 		}
 	}
+	j.mergeMeta(map[string]any{
+		"apply_summary": map[string]any{
+			"revision_id":    revID,
+			"speakers_count": len(applied),
+			"speaker_ids":    applied,
+			"message":        fmt.Sprintf("Ревизия %s применена на %d спикерах", shortID(revID), len(applied)),
+		},
+	})
 	mergeBirdPostApplyMeta(j)
 	j.Succeed()
 }
@@ -210,6 +241,139 @@ func (w *Worker) runRollback(j *Job) {
 		return
 	}
 	j.mergeMeta(map[string]any{"new_revision_id": newID})
+	j.mergeMeta(map[string]any{
+		"rollback_summary": map[string]any{
+			"source_revision_id": src,
+			"new_revision_id":    newID,
+			"message":            fmt.Sprintf("Rollback %s -> %s", shortID(src), shortID(newID)),
+		},
+	})
 	w.enqueueDeployAllSpeakers(j, j.TenantID, newID)
 	j.Succeed()
 }
+
+func (w *Worker) buildRevisionLogEntries(tenantID, revID string) ([]map[string]any, int, error) {
+	if w == nil || w.Store == nil {
+		return nil, 0, fmt.Errorf("store not configured")
+	}
+	var all []store.PrefixRow
+	cursor := ""
+	for {
+		rows, next, more := w.Store.ListRevisionPrefixes(tenantID, revID, cursor, 1000)
+		all = append(all, rows...)
+		if !more {
+			break
+		}
+		cursor = next
+		if strings.TrimSpace(cursor) == "" {
+			break
+		}
+	}
+	type agg struct {
+		kind      string
+		source    string
+		community string
+		count     int
+		sample    []string
+	}
+	groups := map[string]*agg{}
+	for _, p := range all {
+		src := strings.TrimSpace(p.Source)
+		comm := "none"
+		if p.CommunityID != nil && strings.TrimSpace(*p.CommunityID) != "" {
+			comm = strings.TrimSpace(*p.CommunityID)
+		}
+		kind, sourceName := classifySource(src)
+		k := kind + "|" + sourceName + "|" + comm
+		g, ok := groups[k]
+		if !ok {
+			g = &agg{kind: kind, source: sourceName, community: comm}
+			groups[k] = g
+		}
+		g.count++
+		if len(g.sample) < 5 {
+			g.sample = append(g.sample, p.Prefix)
+		}
+	}
+	keys := make([]string, 0, len(groups))
+	for k := range groups {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
+	out := make([]map[string]any, 0, len(keys))
+	for _, k := range keys {
+		g := groups[k]
+		msg := humanLogMessage(g.kind, g.source, g.count, g.community, g.sample)
+		out = append(out, map[string]any{
+			"kind":         g.kind,
+			"source":       g.source,
+			"community":    g.community,
+			"prefix_count": g.count,
+			"sample":       g.sample,
+			"message":      msg,
+		})
+	}
+	return out, len(all), nil
+}
+
+func classifySource(src string) (kind, name string) {
+	switch {
+	case strings.HasPrefix(src, "as:"):
+		return "asn", strings.TrimPrefix(src, "as:")
+	case strings.HasPrefix(src, "domain:"):
+		return "domain", strings.TrimPrefix(src, "domain:")
+	case strings.HasPrefix(src, "cdn:"):
+		return "cdn", strings.TrimPrefix(src, "cdn:")
+	case src == "ip_range":
+		return "ip_range", "manual_ranges"
+	default:
+		if src == "" {
+			return "unknown", "unknown"
+		}
+		return "source", src
+	}
+}
+
+func humanLogMessage(kind, source string, count int, community string, sample []string) string {
+	switch kind {
+	case "asn":
+		return fmt.Sprintf("AS%s -> %d префиксов добавлены в community %s", source, count, community)
+	case "domain":
+		ips := strings.Join(prettyDomainSample(sample), " ")
+		if ips == "" {
+			ips = "-"
+		}
+		return fmt.Sprintf("%s -> ip (%s) -> добавлены в community %s", source, ips, community)
+	case "cdn":
+		return fmt.Sprintf("CDN source %s -> %d префиксов добавлены в community %s", source, count, community)
+	case "ip_range":
+		return fmt.Sprintf("IP ranges -> %d префиксов добавлены в community %s", count, community)
+	default:
+		return fmt.Sprintf("%s -> %d префиксов добавлены в community %s", source, count, community)
+	}
+}
+
+func prettyDomainSample(sample []string) []string {
+	out := make([]string, 0, len(sample))
+	for _, s := range sample {
+		p, err := netip.ParsePrefix(strings.TrimSpace(s))
+		if err != nil {
+			out = append(out, s)
+			continue
+		}
+		if (p.Addr().Is4() && p.Bits() == 32) || (p.Addr().Is6() && p.Bits() == 128) {
+			out = append(out, p.Addr().String())
+			continue
+		}
+		out = append(out, s)
+	}
+	return out
+}
+
+func shortID(id string) string {
+	s := strings.TrimSpace(id)
+	if len(s) <= 8 {
+		return s
+	}
+	return s[:8]
+}