package jobs

import (
	"context"
	"errors"
	"fmt"
	"net/http"
	"net/netip"
	"os"
	"sort"
	"strings"
	"sync"
	"time"

	"evobgp/internal/birddeploy"
	"evobgp/internal/birdfmt"
	"evobgp/internal/httpclient"
	"evobgp/internal/nodedispatch"
	"evobgp/internal/observability"
	"evobgp/internal/pipeline"
	"evobgp/internal/store"

	"github.com/jackc/pgx/v5/pgxpool"
)

// mergeBirdPostApplyMeta attaches a birdc snapshot after deploy/reload (best-effort).
func mergeBirdPostApplyMeta(j *Job) {
	if strings.TrimSpace(os.Getenv("EVOBGP_BIRDC_SOCKET")) == "" {
		j.mergeMeta(map[string]any{"bird_post_apply_check": "skipped_no_birdc_socket"})
		return
	}
	ctx, cancel := context.WithTimeout(context.Background(), 8*time.Second)
	defer cancel()
	st := birdfmt.InspectLocalBird(ctx)
	inner := map[string]any{
		"bgp_established":    st.BGPEstablished,
		"bgp_sessions_total": st.BGPSessionsTotal,
	}
	if st.Error != "" {
		inner["ok"] = false
		inner["error"] = st.Error
	} else {
		inner["ok"] = true
	}
	j.mergeMeta(map[string]any{"bird_post_apply": inner})
}

const (
	KindModuleRefresh          = "module_refresh"
	KindTenantRefresh          = "tenant_refresh"
	KindPeerReconcile          = "peer_reconcile"
	KindDeployApply            = "deploy_apply"
	KindRevisionRollback       = "revision_rollback"
	KindBirdReload             = "bird_reload"
	KindPostgresMetricsRefresh = "postgres_metrics_refresh"
	KindPostgresSlowQueryAgg   = "postgres_slow_query_aggregate"
	KindPostgresTableBloat     = "postgres_table_bloat_estimate"
	KindPostgresIndexUsage     = "postgres_index_usage_analyze"
	KindPostgresAutovacuumLag  = "postgres_autovacuum_lag_detect"
	KindPostgresVacuum         = "postgres_vacuum"
	KindPostgresVacuumAnalyze  = "postgres_vacuum_analyze"
	KindPostgresAnalyze        = "postgres_analyze"
	KindPostgresReindex        = "postgres_reindex"
	KindPostgresCleanup        = "postgres_cleanup"
	KindMaintenancePolicyRun   = "maintenance_policy_run"
)

// Worker executes queued jobs against store.Backend (memory or SQL).
type Worker struct {
	Store          store.Backend
	PgPool         *pgxpool.Pool
	HTTPClient     *http.Client // optional; CDN refresh uses this (default 45s timeout).
	NodeHTTPClient *http.Client // optional; agent dispatch only, never shared with CDN transport.
	// Registry is set after BootstrapWorkers creates the job queue; used to chain deploy_apply after refresh/rollback.
	Registry *Registry
	// refreshGate decides which module refresh renders for a tenant.
	// The mutex covers only that decision; the render itself runs outside the lock.
	refreshGate sync.Map // map[string]*tenantRefreshGate
	// dispatchGate serializes publish and deployment state for one speaker in this process.
	// Agent wake runs outside the mutex.
	dispatchGate sync.Map // map[string]*sync.Mutex
}

type revisionLogEntry struct {
	Kind           string   `json:"kind"`
	Source         string   `json:"source"`
	Community      string   `json:"community"`
	CommunityLabel string   `json:"community_label"`
	PrefixCount    int      `json:"prefix_count"`
	Sample         []string `json:"sample,omitempty"`
	Message        string   `json:"message"`
}

var defaultWorkerHTTP = httpclient.New(httpclient.DefaultTimeout)

func (w *Worker) httpClient() *http.Client {
	if w != nil && w.HTTPClient != nil {
		return w.HTTPClient
	}
	return defaultWorkerHTTP
}

// Process is registered as Registry.workerStart.
func (w *Worker) Process(j *Job) {
	defer func() {
		observability.RecordJobTerminal(j.Kind, j.statusLocked())
		if w != nil && w.Registry != nil {
			w.Registry.fireTerminal(j)
		}
	}()

	if w == nil || w.Store == nil {
		j.MarkRunning()
		j.Fail("worker not configured")
		return
	}
	j.MarkRunning()
	if w != nil && w.Registry != nil {
		w.Registry.fireRunning(j)
	}
	if j.IsCancelRequested() {
		j.MarkCancelled()
		return
	}
	// A committed child is the durable checkpoint for a parent that crashed
	// before its own terminal write. Reuse it before rendering another revision.
	if w.Registry != nil && (isRefreshKind(j.Kind) || j.Kind == KindRevisionRollback || j.Kind == KindPeerReconcile) {
		child, err := w.Registry.FindIdempotentContext(context.Background(), j.TenantID, "deploy_from:"+j.ID)
		if err == nil {
			j.mergeMeta(map[string]any{"revision_id": child.metaString("revision_id"), "deploy_apply_job_id": child.ID, "recovered_child": true})
			if isRefreshKind(j.Kind) {
				w.Registry.finalizeRefreshCoalesce(j.TenantID)
			}
			j.Succeed()
			return
		}
		if !errors.Is(err, store.ErrJobNotFound) {
			j.Fail(err.Error())
			return
		}
	}

	switch j.Kind {
	case KindModuleRefresh:
		w.runModuleRefresh(j)
	case KindTenantRefresh:
		w.runTenantRefresh(j)
	case KindPeerReconcile:
		w.runPeerReconcile(j)
	case KindDeployApply:
		w.runDeployApply(j)
	case KindRevisionRollback:
		w.runRollback(j)
	case KindBirdReload:
		sock := strings.TrimSpace(os.Getenv("EVOBGP_BIRDC_SOCKET"))
		if sock == "" {
			j.Succeed()
			return
		}
		ctx, cancel := j.workContext()
		defer cancel()
		ctl := &birdfmt.BirdCtl{
			Socket: sock,
			Birdc:  strings.TrimSpace(os.Getenv("EVOBGP_BIRDC_BIN")),
		}
		if err := ctl.Configure(ctx); err != nil {
			if ctx.Err() != nil {
				j.MarkCancelled()
				return
			}
			j.Fail(err.Error())
			return
		}
		mergeBirdPostApplyMeta(j)
		j.Succeed()
	case KindPostgresMetricsRefresh:
		w.runPostgresMetricsRefresh(j)
	case KindPostgresSlowQueryAgg:
		w.runPostgresSlowQueryAgg(j)
	case KindPostgresTableBloat:
		w.runPostgresTableBloat(j)
	case KindPostgresIndexUsage:
		w.runPostgresIndexUsage(j)
	case KindPostgresAutovacuumLag:
		w.runPostgresAutovacuumLag(j)
	case KindPostgresVacuum:
		w.runPostgresMaint(j, "vacuum")
	case KindPostgresVacuumAnalyze:
		w.runPostgresMaint(j, "vacuum_analyze")
	case KindPostgresAnalyze:
		w.runPostgresMaint(j, "analyze")
	case KindPostgresReindex:
		w.runPostgresMaint(j, "reindex")
	case KindPostgresCleanup:
		w.runPostgresCleanup(j)
	case KindMaintenancePolicyRun:
		w.runMaintenancePolicy(j)
	default:
		j.Fail("unknown job kind")
	}
}

func (w *Worker) runPeerReconcile(j *Job) {
	if w == nil || w.Store == nil {
		j.Fail("worker not configured")
		return
	}
	const peerJobTitle = "Обновление BGP пиров"
	j.mergeMeta(map[string]any{"job_title": peerJobTitle})

	latest, _, _ := w.Store.ListRevisions(j.TenantID, "", "", 1)
	triggerModuleID, err := w.peerTriggerModuleID(j.TenantID, latest)
	if err != nil {
		j.Fail(err.Error())
		return
	}
	// Peer edits bump render generation. Render from module snapshots so the
	// new revision includes current prefixes without paging the previous one.
	ctx, cancel := j.workContext()
	defer cancel()
	revID, err := pipeline.RenderTenantRevision(ctx, w.Store, w.httpClient(), j.TenantID, triggerModuleID)
	if err != nil {
		if ctx.Err() != nil {
			j.MarkCancelled()
			return
		}
		j.Fail(err.Error())
		return
	}
	j.mergeMeta(map[string]any{"revision_id": revID})
	if entries, total, err := w.buildRevisionLogEntries(j.TenantID, revID); 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()})
	}
	if err := w.enqueueDeployAllSpeakers(j, j.TenantID, revID); err != nil {
		j.Fail(err.Error())
		return
	}
	j.Succeed()
}

func (w *Worker) peerTriggerModuleID(tenantID string, latest []*store.Revision) (string, error) {
	if len(latest) > 0 {
		if mid := strings.TrimSpace(latest[0].ModuleID); mid != "" {
			return mid, nil
		}
	}
	for _, mod := range w.Store.ListModules(tenantID) {
		if mod == nil || !mod.Enabled {
			continue
		}
		if strings.TrimSpace(mod.ID) != "" {
			return mod.ID, nil
		}
	}
	for _, mod := range w.Store.ListModules(tenantID) {
		if mod == nil {
			continue
		}
		if strings.TrimSpace(mod.ID) != "" {
			return mod.ID, nil
		}
	}
	return "", fmt.Errorf("missing module_id for peer reconcile")
}

type tenantRefreshGate struct {
	mu        sync.Mutex
	rendering bool
	dirty     bool
}

func (w *Worker) tenantRefreshGate(tenantID string) *tenantRefreshGate {
	v, _ := w.refreshGate.LoadOrStore(tenantID, &tenantRefreshGate{})
	return v.(*tenantRefreshGate)
}

func (w *Worker) tenantRefreshMu(tenantID string) *sync.Mutex {
	return &w.tenantRefreshGate(tenantID).mu
}

func (w *Worker) speakerDispatchMu(speakerID string) *sync.Mutex {
	v, _ := w.dispatchGate.LoadOrStore(speakerID, &sync.Mutex{})
	return v.(*sync.Mutex)
}

// finishModuleRefreshSuccess marks the refresh job and, for the last active refresh in tenant,
// creates one aggregate revision and enqueues a single deploy_apply.
// runModuleRefresh handles a single module_refresh job and guarantees the per-tenant inflight
// slot is released exactly once — even on failure/cancellation before finishModuleRefreshSuccess.
func (w *Worker) runModuleRefresh(j *Job) {
	coalesceFinalized := false
	defer func() {
		if !coalesceFinalized && w != nil && w.Registry != nil {
			// Refresh failed/was cancelled before reaching finishModuleRefreshSuccess.
			// Decrement the counter under the tenant mutex so the "last one" logic stays sound.
			mu := w.tenantRefreshMu(j.TenantID)
			mu.Lock()
			w.Registry.finalizeRefreshCoalesce(j.TenantID)
			mu.Unlock()
		}
	}()

	mid := strings.TrimSpace(j.metaString("module_id"))
	if mid == "" {
		j.Fail("missing module_id in job meta")
		return
	}
	ctx, cancel := j.workContext()
	defer cancel()
	if ctx.Err() != nil {
		j.MarkCancelled()
		return
	}
	if err := pipeline.RefreshModuleIngest(ctx, w.Store, w.httpClient(), j.TenantID, mid); err != nil {
		if ctx.Err() != nil {
			j.MarkCancelled()
			return
		}
		j.Fail(err.Error())
		return
	}
	w.finishModuleRefreshSuccess(j, mid)
	coalesceFinalized = true
}

func (w *Worker) runTenantRefresh(j *Job) {
	coalesceFinalized := false
	defer func() {
		if !coalesceFinalized && w != nil && w.Registry != nil {
			mu := w.tenantRefreshMu(j.TenantID)
			mu.Lock()
			w.Registry.finalizeRefreshCoalesce(j.TenantID)
			mu.Unlock()
		}
	}()

	moduleIDs := moduleIDsFromJobMeta(j.metaCopy())
	if len(moduleIDs) == 0 {
		j.mergeMeta(map[string]any{"modules_refreshed": 0})
		j.Succeed()
		return
	}
	ctx, cancel := j.workContext()
	defer cancel()
	if err := pipeline.RefreshTenantModules(ctx, w.Store, w.httpClient(), j.TenantID, moduleIDs); err != nil {
		if ctx.Err() != nil {
			j.MarkCancelled()
			return
		}
		j.Fail(err.Error())
		return
	}
	trigger, err := pipeline.PickTenantRefreshTriggerModule(w.Store, j.TenantID, moduleIDs)
	if err != nil {
		j.Fail(err.Error())
		return
	}
	j.mergeMeta(map[string]any{"module_ids": moduleIDs, "modules_refreshed": len(moduleIDs)})
	w.finishModuleRefreshSuccess(j, trigger)
	coalesceFinalized = true
}

func moduleIDsFromJobMeta(meta map[string]any) []string {
	if meta == nil {
		return nil
	}
	raw, ok := meta["module_ids"]
	if !ok {
		return nil
	}
	switch v := raw.(type) {
	case []string:
		return v
	case []any:
		var out []string
		for _, x := range v {
			if s, ok := x.(string); ok && strings.TrimSpace(s) != "" {
				out = append(out, strings.TrimSpace(s))
			}
		}
		return out
	default:
		return nil
	}
}

func (w *Worker) finishModuleRefreshSuccess(j *Job, triggerModuleID string) {
	if w == nil || w.Store == nil {
		if w != nil && w.Registry != nil {
			w.Registry.finalizeRefreshCoalesce(j.TenantID)
		}
		j.Succeed()
		return
	}
	gate := w.tenantRefreshGate(j.TenantID)
	gate.mu.Lock()
	// The counter is incremented in Enqueue and decremented here, so the "last one"
	// decision does not depend on job status polling. Render runs after the unlock.
	isLastRefresh := true
	if w.Registry != nil {
		isLastRefresh = w.Registry.finalizeRefreshCoalesce(j.TenantID)
	}
	if !isLastRefresh || gate.rendering {
		if gate.rendering {
			gate.dirty = true
		}
		gate.mu.Unlock()
		j.mergeMeta(map[string]any{
			"deploy_apply_deferred":     true,
			"deploy_apply_defer_reason": "parallel_module_refresh",
		})
		j.Succeed()
		return
	}
	gate.rendering = true
	gate.mu.Unlock()
	defer func() {
		gate.mu.Lock()
		gate.rendering = false
		gate.mu.Unlock()
	}()

	ctx, cancel := j.workContext()
	defer cancel()
	rev, err := w.renderTenantUntilQuiet(ctx, j, triggerModuleID)
	if err != nil {
		if errors.Is(err, errRenderSuperseded) {
			j.Fail(err.Error())
			return
		}
		if ctx.Err() != nil || j.IsCancelRequested() {
			j.MarkCancelled()
			return
		}
		j.Fail(err.Error())
		return
	}
	if ctx.Err() != nil || j.IsCancelRequested() {
		j.MarkCancelled()
		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()})
	}
	if err := w.enqueueDeployAllSpeakers(j, j.TenantID, rev); err != nil {
		j.Fail(err.Error())
		return
	}
	j.Succeed()
}

const (
	// renderQuietPassLimit stops a render loop that never observes a quiet input set.
	// Hitting the cap fails the job; the last revision must not be published.
	renderQuietPassLimit = 32
	// speakerWakeLimit is the maximum number of concurrent agent wakes in one deploy.
	speakerWakeLimit = 4
)

// errRenderSuperseded marks a render that must not be deployed. A newer snapshot
// landed, the job was cancelled while inputs were still changing, or the quiet
// pass cap was exhausted.
var errRenderSuperseded = errors.New("newer snapshot still pending")

// renderTenantUntilQuiet builds a revision outside the tenant gate.
// It keeps rendering while another ingest marks the gate dirty or the snapshot
// generation moves, and returns a revision only after a fresh check shows the
// inputs are still the ones just rendered. Cancellation or the safety cap while
// inputs are still changing fails the render instead of publishing a stale revision.
func (w *Worker) renderTenantUntilQuiet(ctx context.Context, j *Job, triggerModuleID string) (string, error) {
	gate := w.tenantRefreshGate(j.TenantID)
	var rev string
	for pass := 0; pass < renderQuietPassLimit; pass++ {
		if abort := renderJobAborted(ctx, j); abort != nil {
			if takeRefreshDirty(gate) {
				return "", errRenderWhileDirty(abort)
			}
			return "", abort
		}
		before, err := w.captureRenderInputs(ctx, j.TenantID)
		if err != nil {
			if abort := renderJobAborted(ctx, j); abort != nil && takeRefreshDirty(gate) {
				return "", errRenderWhileDirty(abort)
			}
			return "", err
		}
		rev, err = pipeline.RenderTenantRevision(ctx, w.Store, w.httpClient(), j.TenantID, triggerModuleID)
		if err != nil {
			dirty := takeRefreshDirty(gate)
			if abort := renderJobAborted(ctx, j); abort != nil && dirty {
				return "", errRenderWhileDirty(abort)
			}
			if abort := renderJobAborted(ctx, j); abort != nil {
				return "", abort
			}
			if dirty {
				continue
			}
			return "", err
		}
		after, err := w.captureRenderInputs(ctx, j.TenantID)
		if err != nil {
			if abort := renderJobAborted(ctx, j); abort != nil && takeRefreshDirty(gate) {
				return "", errRenderWhileDirty(abort)
			}
			return "", err
		}
		dirty := takeRefreshDirty(gate)
		if dirty || renderInputsChanged(before, after) {
			if abort := renderJobAborted(ctx, j); abort != nil {
				return "", errRenderWhileDirty(abort)
			}
			continue
		}
		confirm, err := w.captureRenderInputs(ctx, j.TenantID)
		if err != nil {
			if abort := renderJobAborted(ctx, j); abort != nil && takeRefreshDirty(gate) {
				return "", errRenderWhileDirty(abort)
			}
			return "", err
		}
		if takeRefreshDirty(gate) || renderInputsChanged(after, confirm) {
			if abort := renderJobAborted(ctx, j); abort != nil {
				return "", errRenderWhileDirty(abort)
			}
			continue
		}
		if abort := renderJobAborted(ctx, j); abort != nil {
			return "", abort
		}
		return rev, nil
	}
	if abort := renderJobAborted(ctx, j); abort != nil {
		return "", errRenderWhileDirty(abort)
	}
	return "", errRenderWhileDirty(nil)
}

func errRenderWhileDirty(cause error) error {
	if cause == nil {
		return fmt.Errorf("jobs: tenant render did not quiesce after %d passes: %w", renderQuietPassLimit, errRenderSuperseded)
	}
	return fmt.Errorf("jobs: tenant render cancelled while a newer snapshot was still pending: %w", errRenderSuperseded)
}

func renderJobAborted(ctx context.Context, j *Job) error {
	if err := ctx.Err(); err != nil {
		return err
	}
	if j != nil && j.IsCancelRequested() {
		return context.Canceled
	}
	return nil
}

func takeRefreshDirty(gate *tenantRefreshGate) bool {
	if gate == nil {
		return false
	}
	gate.mu.Lock()
	defer gate.mu.Unlock()
	dirty := gate.dirty
	gate.dirty = false
	return dirty
}

type renderInputView struct {
	generation int64
	snapshots  []store.ModuleSnapshotUse
}

func (w *Worker) captureRenderInputs(ctx context.Context, tenantID string) (renderInputView, error) {
	if w == nil || w.Store == nil {
		return renderInputView{}, fmt.Errorf("jobs: render store unavailable")
	}
	gen, err := w.Store.GetRenderGeneration(ctx, tenantID)
	if err != nil {
		if !errors.Is(err, store.ErrNotFound) {
			return renderInputView{}, fmt.Errorf("jobs: read render generation: %w", err)
		}
		gen = 0
	}
	headers, err := w.Store.ListModuleSnapshotHeaders(ctx, tenantID)
	if err != nil {
		return renderInputView{}, fmt.Errorf("jobs: read snapshot headers: %w", err)
	}
	return renderInputView{generation: gen, snapshots: headers}, nil
}

func renderInputsChanged(before, after renderInputView) bool {
	if before.generation != after.generation {
		return true
	}
	return !sameSnapshotHeaders(before.snapshots, after.snapshots)
}

func sameSnapshotHeaders(a, b []store.ModuleSnapshotUse) bool {
	if len(a) != len(b) {
		return false
	}
	aa := append([]store.ModuleSnapshotUse(nil), a...)
	bb := append([]store.ModuleSnapshotUse(nil), b...)
	sort.Slice(aa, func(i, j int) bool { return aa[i].ModuleID < aa[j].ModuleID })
	sort.Slice(bb, func(i, j int) bool { return bb[i].ModuleID < bb[j].ModuleID })
	for i := range aa {
		if aa[i].ModuleID != bb[i].ModuleID || aa[i].SnapshotVersion != bb[i].SnapshotVersion || aa[i].InputHash != bb[i].InputHash {
			return false
		}
	}
	return true
}

// enqueueDeployAllSpeakers queues the same work as POST /v1/apply (all speakers, no speaker_id).
func (w *Worker) enqueueDeployAllSpeakers(j *Job, tenantID, revID string) error {
	if w == nil || w.Registry == nil {
		return fmt.Errorf("jobs: deploy queue unavailable")
	}
	revID = strings.TrimSpace(revID)
	if revID == "" {
		return fmt.Errorf("jobs: missing revision for deploy")
	}
	meta := map[string]any{"revision_id": revID}
	if tr := strings.TrimSpace(j.metaString("trigger")); tr != "" {
		meta["trigger"] = tr
	}
	key := "deploy_from:" + j.ID
	applyJob, _, err := w.Registry.Enqueue(tenantID, KindDeployApply, &key, nil, meta)
	if err != nil {
		j.mergeMeta(map[string]any{"deploy_apply_enqueue_error": err.Error()})
		return fmt.Errorf("jobs: enqueue deploy: %w", err)
	}
	if applyJob != nil {
		j.mergeMeta(map[string]any{"deploy_apply_job_id": applyJob.ID})
	}
	return nil
}

func (w *Worker) runDeployApply(j *Job) {
	revID := strings.TrimSpace(j.metaString("revision_id"))
	spk := strings.TrimSpace(j.metaString("speaker_id"))
	if revID == "" {
		j.Fail("missing revision_id in job meta")
		return
	}
	if j.OwnerID == "" {
		if _, developmentMemory := w.Store.(*store.Memory); !developmentMemory {
			j.Fail("deployment requires a current durable job lease")
			return
		}
	}

	ctx, cancel := j.workContext()
	defer cancel()
	cancelled := func() bool { return ctx.Err() != nil || j.IsCancelRequested() }
	var speakers []*store.Speaker
	if spk != "" {
		sp, err := w.Store.GetSpeaker(j.TenantID, spk)
		if err != nil {
			j.Fail(err.Error())
			return
		}
		speakers = []*store.Speaker{sp}
	} else {
		speakers = w.Store.ListSpeakersForTenant(j.TenantID)
	}
	activeDir := strings.TrimSpace(os.Getenv("EVOBGP_BIRD_ACTIVE_DIR"))
	localAttempted := false
	var localErr error
	applyLocal := func() error {
		if localAttempted {
			return localErr
		}
		localAttempted = true
		revObj, err := w.Store.GetRevision(j.TenantID, revID)
		if err != nil {
			localErr = err
			return err
		}
		staging := strings.TrimSpace(os.Getenv("EVOBGP_BIRD_STAGING_DIR"))
		if staging == "" {
			staging = os.TempDir() + "/evobgp-bird-staging"
		}
		cfg := birddeploy.Config{
			ActiveDir:  activeDir,
			StagingDir: staging,
			BirdBin:    strings.TrimSpace(os.Getenv("EVOBGP_BIRD_BIN")),
			BirdcBin:   strings.TrimSpace(os.Getenv("EVOBGP_BIRDC_BIN")),
			Socket:     strings.TrimSpace(os.Getenv("EVOBGP_BIRDC_SOCKET")),
		}
		ctl := &birdfmt.BirdCtl{Bird: cfg.BirdBin, Birdc: cfg.BirdcBin, Socket: cfg.Socket}
		if err := birddeploy.ApplyRevision(ctx, ctl, revObj, cfg); err != nil {
			localErr = err
			return err
		}
		return nil
	}
	tally := &deployTally{applied: make([]string, 0, len(speakers))}
	slots := make(chan struct{}, speakerWakeLimit)
	var wakeGroup sync.WaitGroup
	acquireWake := func() bool {
		select {
		case slots <- struct{}{}:
		case <-ctx.Done():
			return false
		}
		if cancelled() {
			<-slots
			return false
		}
		return true
	}
	for _, sp := range speakers {
		if cancelled() {
			break
		}
		if sp == nil {
			continue
		}
		held := false
		if speakerRemoteWake(sp) {
			if !acquireWake() {
				break
			}
			held = true
		}
		staged := w.stageSpeakerDeploy(ctx, j, sp, revID, activeDir, cancelled, applyLocal)
		if staged.wake && !staged.stop {
			wakeGroup.Add(1)
			go func(sp *store.Speaker, deployment store.DeploymentSpeaker) {
				defer wakeGroup.Done()
				defer func() { <-slots }()
				wakeCtx, cancelWake := context.WithTimeout(ctx, 35*time.Second)
				res := nodedispatch.WakeSpeaker(wakeCtx, sp, nodedispatch.Options{RevisionID: revID, HTTPClient: w.NodeHTTPClient, MaxRetries: 1})
				cancelWake()
				out, include, failedN, unconfirmedN, appliedID := w.finishSpeakerWake(ctx, j, sp, revID, deployment, res, cancelled)
				tally.add(out, include, 0, failedN, unconfirmedN, appliedID)
			}(sp, staged.deployment)
		} else if held {
			<-slots
		}
		tally.add(staged.result, staged.hasResult, staged.published, staged.failed, staged.unconfirmed, staged.appliedID)
		if staged.stop {
			break
		}
	}
	wakeGroup.Wait()
	results, applied, published, failed, unconfirmed := tally.snapshot()
	j.mergeMeta(map[string]any{
		"apply_summary": map[string]any{
			"revision_id":       revID,
			"speakers_count":    len(speakers),
			"published_count":   published,
			"applied_count":     len(applied),
			"unconfirmed_count": unconfirmed,
			"failed_count":      failed,
			"speaker_ids":       applied,
			"message":           fmt.Sprintf("Ревизия %s: подтверждено %d, без подтверждения %d, ошибок %d", shortID(revID), len(applied), unconfirmed, failed),
		},
		"node_dispatch": map[string]any{"revision_id": revID, "results": results},
	})
	if cancelled() {
		j.MarkCancelled()
		return
	}
	mergeBirdPostApplyMeta(j)
	if cancelled() {
		j.MarkCancelled()
		return
	}
	if failed > 0 {
		j.Fail(fmt.Sprintf("deploy delivery failed for %d of %d speakers", failed, len(speakers)))
		return
	}
	j.Succeed()
}

type speakerStage struct {
	stop        bool
	wake        bool
	deployment  store.DeploymentSpeaker
	result      nodedispatch.Result
	hasResult   bool
	published   int
	failed      int
	unconfirmed int
	appliedID   string
}

type deployTally struct {
	mu          sync.Mutex
	results     []nodedispatch.Result
	applied     []string
	published   int
	failed      int
	unconfirmed int
}

func (t *deployTally) add(res nodedispatch.Result, include bool, published, failed, unconfirmed int, appliedID string) {
	t.mu.Lock()
	defer t.mu.Unlock()
	if include {
		t.results = append(t.results, res)
	}
	t.published += published
	t.failed += failed
	t.unconfirmed += unconfirmed
	if appliedID != "" {
		t.applied = append(t.applied, appliedID)
	}
}

func (t *deployTally) snapshot() (results []nodedispatch.Result, applied []string, published, failed, unconfirmed int) {
	t.mu.Lock()
	defer t.mu.Unlock()
	results = append([]nodedispatch.Result(nil), t.results...)
	applied = append([]string(nil), t.applied...)
	if applied == nil {
		applied = []string{}
	}
	sort.Slice(results, func(i, j int) bool { return results[i].SpeakerID < results[j].SpeakerID })
	sort.Strings(applied)
	return results, applied, t.published, t.failed, t.unconfirmed
}

func speakerRemoteWake(sp *store.Speaker) bool {
	if sp == nil || !nodedispatch.Enabled() {
		return false
	}
	return store.SpeakerNeedsRemoteDispatch(sp.Role, store.ParseSpeakerMeta(sp.MetaJSON))
}

func (w *Worker) bindDeploymentWriters(ctx context.Context, j *Job, sp *store.Speaker, revID string, deployment store.DeploymentSpeaker) (func(string, string, string) error, func(string) error) {
	fence := deploymentFenceForJob(j)
	record := func(state, confirmation, message string) error {
		if j.OwnerID == "" {
			return nil
		}
		return w.Store.RecordDeploymentSpeaker(ctx, fence, deployment, state, confirmation, message)
	}
	markApplied := func(confirmation string) error {
		if j.OwnerID == "" {
			return w.Store.SetLastAppliedRevision(j.TenantID, sp.ID, revID)
		}
		return record("applied", confirmation, "")
	}
	return record, markApplied
}

// stageSpeakerDeploy publishes one speaker and records terminal states.
// Remote wakes are left to the caller so the speaker lock is not held across the network.
func (w *Worker) stageSpeakerDeploy(ctx context.Context, j *Job, sp *store.Speaker, revID, activeDir string, cancelled func() bool, applyLocal func() error) speakerStage {
	var staged speakerStage
	mu := w.speakerDispatchMu(sp.ID)
	mu.Lock()
	defer mu.Unlock()
	if cancelled() {
		staged.stop = true
		return staged
	}
	var deployment store.DeploymentSpeaker
	var publishErr error
	if j.OwnerID != "" {
		deployment, publishErr = beginDeployment(ctx, w.Store, deploymentFenceForJob(j), sp.ID, revID)
	} else {
		publishErr = w.Store.PublishRevisionForSpeaker(sp.ID, revID)
	}
	staged.deployment = deployment
	record, markApplied := w.bindDeploymentWriters(ctx, j, sp, revID, deployment)
	if publishErr != nil {
		staged.failed = 1
		staged.hasResult = true
		staged.result = nodedispatch.Result{SpeakerID: sp.ID, Status: "error", Error: publishErr.Error()}
		return staged
	}
	staged.published = 1
	if deployment.State == "applied" {
		staged.appliedID = sp.ID
		staged.hasResult = true
		staged.result = nodedispatch.Result{SpeakerID: sp.ID, Status: "ok", Confirmation: deployment.Confirmation, AppliedRevisionID: revID}
		return staged
	}
	if cancelled() {
		staged.stop = true
		return staged
	}
	if !speakerRemoteWake(sp) {
		if activeDir != "" && strings.EqualFold(sp.Role, "master") {
			err := applyLocal()
			if cancelled() {
				staged.stop = true
				return staged
			}
			if err != nil {
				_ = record("failed", "unconfirmed", err.Error())
				staged.failed = 1
				staged.hasResult = true
				staged.result = nodedispatch.Result{SpeakerID: sp.ID, Status: "error", Error: err.Error()}
				return staged
			}
			if err := markApplied("local_applied"); err != nil {
				staged.failed = 1
				staged.hasResult = true
				staged.result = nodedispatch.Result{SpeakerID: sp.ID, Status: "error", Error: err.Error()}
				return staged
			}
			staged.appliedID = sp.ID
			staged.hasResult = true
			staged.result = nodedispatch.Result{SpeakerID: sp.ID, Status: "ok", Confirmation: "local_applied", AppliedRevisionID: revID}
			return staged
		}
		if err := record("unconfirmed", "unconfirmed", "dispatch disabled or speaker has no remote agent"); err != nil {
			staged.failed = 1
		}
		staged.unconfirmed = 1
		staged.hasResult = true
		staged.result = nodedispatch.Result{SpeakerID: sp.ID, Status: "skipped", Confirmation: "unconfirmed"}
		return staged
	}
	if err := record("dispatched", "unconfirmed", "awaiting agent acknowledgement"); err != nil {
		staged.failed = 1
		staged.hasResult = true
		staged.result = nodedispatch.Result{SpeakerID: sp.ID, Status: "error", Error: err.Error()}
		return staged
	}
	if deployment.DispatchAttempts >= 3 {
		_ = record("unknown", "unconfirmed", "delivery attempt budget exhausted")
		staged.failed = 1
		staged.hasResult = true
		staged.result = nodedispatch.Result{SpeakerID: sp.ID, Status: "error", Confirmation: "unconfirmed", Error: "delivery attempt budget exhausted"}
		return staged
	}
	staged.wake = true
	return staged
}

// finishSpeakerWake records the agent acknowledgement. The speaker lock covers the
// state update and is not held while WakeSpeaker is in flight.
func (w *Worker) finishSpeakerWake(ctx context.Context, j *Job, sp *store.Speaker, revID string, deployment store.DeploymentSpeaker, res nodedispatch.Result, cancelled func() bool) (nodedispatch.Result, bool, int, int, string) {
	mu := w.speakerDispatchMu(sp.ID)
	mu.Lock()
	defer mu.Unlock()
	if cancelled != nil && cancelled() {
		return nodedispatch.Result{SpeakerID: sp.ID, Endpoint: res.Endpoint, Status: "cancelled", Confirmation: "unconfirmed"}, true, 0, 0, ""
	}
	record, markApplied := w.bindDeploymentWriters(ctx, j, sp, revID, deployment)
	appliedID := ""
	if res.Status == "ok" && res.AppliedRevisionID != "" {
		if res.AppliedRevisionID != revID {
			res.Status = "error"
			res.Confirmation = "unconfirmed"
			res.Error = "agent reported a different applied revision"
		} else if latest, _, err := w.Store.LatestPublishedRevision(sp.ID); err != nil {
			res.Status = "error"
			res.Confirmation = "unconfirmed"
			res.Error = err.Error()
		} else if latest != revID {
			res.Status = "error"
			res.Confirmation = "unconfirmed"
			res.Error = "stale agent acknowledgement: desired revision changed"
		} else if err := markApplied("agent_reported"); err != nil {
			res.Status = "error"
			res.Confirmation = "unconfirmed"
			res.Error = err.Error()
		} else {
			appliedID = sp.ID
		}
	}
	if res.Status != "ok" || res.Confirmation != "agent_reported" {
		state := "unconfirmed"
		if res.Status != "ok" {
			state = "unknown"
		}
		if err := record(state, "unconfirmed", res.Error); err != nil {
			res.Status = "error"
			res.Error = err.Error()
		}
	}
	patch := store.SpeakerMeta{
		LastDispatchAt:     time.Now().UTC().Format(time.RFC3339Nano),
		LastDispatchStatus: res.Status,
	}
	if res.Error != "" {
		patch.LastDispatchError = res.Error
		patch.SyncStatus = "error"
	} else if res.Status == "ok" && res.Confirmation == "agent_reported" {
		patch.LastDispatchError = ""
		patch.SyncStatus = "synced"
	} else if res.Status == "ok" {
		patch.SyncStatus = "unconfirmed"
	}
	merged := store.MergeSpeakerMetaJSON(sp.MetaJSON, patch)
	if j.OwnerID == "" {
		if _, err := w.Store.UpdateSpeaker(j.TenantID, sp.ID, &store.SpeakerPatch{MetaJSON: &merged}); err != nil {
			res.Status = "error"
			res.Error = "dispatch metadata update: " + err.Error()
		}
	}
	failed := 0
	unconfirmed := 0
	if res.Status == "ok" && res.Confirmation == "unconfirmed" {
		unconfirmed = 1
	} else if res.Status != "ok" {
		failed = 1
	}
	return res, true, failed, unconfirmed, appliedID
}

func (w *Worker) runRollback(j *Job) {
	src := strings.TrimSpace(j.metaString("source_revision_id"))
	if src == "" {
		j.Fail("missing source_revision_id in job meta")
		return
	}
	newID, err := w.Store.CreateRollbackRevision(j.TenantID, src)
	if err != nil {
		j.Fail(err.Error())
		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("Откат %s → %s", shortID(src), shortID(newID)),
		},
	})
	if err := w.enqueueDeployAllSpeakers(j, j.TenantID, newID); err != nil {
		j.Fail(err.Error())
		return
	}
	j.Succeed()
}

// buildCommunityLabelMap maps community UUID -> human-readable title (or BGP community string).
func buildCommunityLabelMap(st store.Backend, tenantID string) map[string]string {
	out := make(map[string]string)
	if st == nil {
		return out
	}
	list, err := st.ListCommunities(tenantID)
	if err != nil || list == nil {
		return out
	}
	for _, c := range list {
		if c == nil {
			continue
		}
		label := strings.TrimSpace(c.Title)
		if label == "" {
			label = strings.TrimSpace(c.Community)
		}
		if label == "" {
			label = c.ID
		}
		out[c.ID] = label
	}
	return out
}

func resolveCommunityLabel(commID string, byID map[string]string) string {
	if commID == "" || commID == "none" {
		return "без community"
	}
	if lbl, ok := byID[commID]; ok && strings.TrimSpace(lbl) != "" {
		return strings.TrimSpace(lbl)
	}
	return commID
}

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")
	}
	commLabels := buildCommunityLabelMap(w.Store, tenantID)
	type agg struct {
		kind      string
		source    string
		community string
		count     int
		sample    []string
	}
	groups := map[string]*agg{}
	total := 0
	cursor := ""
	for {
		rows, next, more := w.Store.ListRevisionPrefixes(tenantID, revID, cursor, 1000)
		for _, p := range rows {
			total++
			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)
			}
		}
		if !more {
			break
		}
		cursor = next
		if strings.TrimSpace(cursor) == "" {
			break
		}
	}
	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]
		cl := resolveCommunityLabel(g.community, commLabels)
		msg := humanLogMessage(g.kind, g.source, g.count, cl, g.sample)
		out = append(out, map[string]any{
			"kind":            g.kind,
			"source":          g.source,
			"community":       g.community,
			"community_label": cl,
			"prefix_count":    g.count,
			"sample":          g.sample,
			"message":         msg,
		})
	}
	return out, total, 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
	}
}

// humanLogMessage builds a Russian log line; communityLabel is already resolved (title or BGP value).
func humanLogMessage(kind, source string, count int, communityLabel string, sample []string) string {
	switch kind {
	case "asn":
		return fmt.Sprintf("AS%s: добавлено %d префиксов в сообщество «%s»", source, count, communityLabel)
	case "domain":
		ips := strings.Join(prettyDomainSample(sample), " ")
		if ips == "" {
			ips = "—"
		}
		return fmt.Sprintf("%s: IP (%s) → добавлено в сообщество «%s»", source, ips, communityLabel)
	case "cdn":
		return fmt.Sprintf("CDN «%s»: добавлено %d префиксов в сообщество «%s»", source, count, communityLabel)
	case "ip_range":
		return fmt.Sprintf("Статические диапазоны: добавлено %d префиксов в сообщество «%s»", count, communityLabel)
	default:
		return fmt.Sprintf("%s: добавлено %d префиксов в сообщество «%s»", source, count, communityLabel)
	}
}

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]
}