EvoBGP/internal/scheduler/run.go

// Package scheduler drives module refresh intervals and enqueues tenant_refresh jobs on the shared Registry.
package scheduler

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"net/http"
	"strings"
	"time"

	"evobgp/internal/broker"
	"evobgp/internal/config"
	"evobgp/internal/httpclient"
	"evobgp/internal/jobs"
	"evobgp/internal/logging"
	"evobgp/internal/pipeline"
	"evobgp/internal/store"
)

// Deps wires the scheduler to a trusted durable Registry (or local evobgp-all),
// or to a single tenant over HTTP using APIBase, APIToken and APITenantID.
type Deps struct {
	Store       store.Backend
	Jobs        *jobs.Registry
	APIBase     string // e.g. http://evobgp-api:8080
	APIToken    string // Tenant-scoped bearer token (operator/editor)
	APITenantID string // Explicit single tenant for HTTP mode; never enumerate other tenants
	HTTP        *http.Client
}

// Run blocks until ctx is cancelled. Misconfigured deps terminate the process (no idle fallback).
func Run(ctx context.Context, deps *Deps) {
	cfg := config.Load()
	broker.LogConnect(ctx, cfg.BrokerURL)
	if deps == nil || deps.Store == nil {
		log.Fatalf("evobgp-scheduler: missing store (pass scheduler.Deps with Store from BootstrapWorkers or evobgp-all)")
	}
	if deps.Jobs == nil && (strings.TrimSpace(deps.APIBase) == "" || strings.TrimSpace(deps.APIToken) == "" || strings.TrimSpace(deps.APITenantID) == "") {
		log.Fatalf("evobgp-scheduler: need either in-process Jobs (evobgp-all) or EVOBGP_CONTROL_PLANE_URL, EVOBGP_SCHEDULER_BEARER and EVOBGP_SCHEDULER_TENANT_ID")
	}
	t := time.NewTicker(time.Duration(pipeline.SchedulerTickSec) * time.Second)
	defer t.Stop()
	if deps.Jobs != nil {
		logging.Default().Info(fmt.Sprintf("evobgp-scheduler: active (registry enqueue tenant_refresh)"))
	} else {
		logging.Default().Info(fmt.Sprintf("evobgp-scheduler: active (HTTP POST .../tenant/refresh → %s)", strings.TrimSpace(deps.APIBase)))
	}
	for {
		select {
		case <-ctx.Done():
			logging.Default().Info(fmt.Sprintf("evobgp-scheduler: stopped"))
			return
		case <-t.C:
			tick(ctx, deps)
		}
	}
}

func tick(ctx context.Context, deps *Deps) {
	tickAt(ctx, deps, time.Now().UTC())
}

func tickAt(ctx context.Context, deps *Deps, now time.Time) {
	ctx, cancel := context.WithTimeout(ctx, 45*time.Second)
	defer cancel()
	if ctx.Err() != nil {
		return
	}
	var tenants []string
	if deps.Jobs == nil {
		tenant := strings.TrimSpace(deps.APITenantID)
		if tenant == "" {
			return
		}
		tenants = []string{tenant}
	} else {
		var err error
		tenants, err = deps.Store.ListTenantIDs()
		if err != nil {
			logging.Default().Error("scheduler list tenants", "err", err)
			return
		}
	}
	tickBucket := now.Unix() / pipeline.SchedulerTickSec
	for _, tid := range tenants {
		if ctx.Err() != nil {
			return
		}
		if tid == store.SystemJobTenantID {
			continue
		}
		var due []string
		for _, mod := range deps.Store.ListModules(tid) {
			if pipeline.ModuleDueForScheduler(mod, now) {
				due = append(due, mod.ID)
			}
		}
		if len(due) == 0 {
			continue
		}
		key := fmt.Sprintf("sched-tenant-%s-%d", tid, tickBucket)
		if deps.Jobs != nil {
			ids := append([]string(nil), due...)
			_, created, err := deps.Jobs.EnqueueContext(ctx, tid, jobs.KindTenantRefresh, &key, nil, map[string]any{
				"module_ids": ids,
				"trigger":    "scheduler",
			})
			if err != nil {
				logging.Default().Info(fmt.Sprintf("evobgp-scheduler: enqueue tenant %s: %v", tid, err))
				continue
			}
			if created {
				logging.Default().Info(fmt.Sprintf("evobgp-scheduler: queued tenant refresh for %d module(s) in tenant %s", len(due), tid))
			}
			continue
		}
		if err := postTenantRefresh(ctx, deps, due, key); err != nil {
			logging.Default().Info(fmt.Sprintf("evobgp-scheduler: http tenant refresh %s: %v", tid, err))
		}
	}
}

func postTenantRefresh(ctx context.Context, deps *Deps, moduleIDs []string, idempotencyKey string) error {
	base := strings.TrimRight(strings.TrimSpace(deps.APIBase), "/")
	u := base + "/v1/tenant/refresh"
	body, err := json.Marshal(map[string]any{"module_ids": moduleIDs})
	if err != nil {
		return err
	}
	req, err := http.NewRequestWithContext(ctx, http.MethodPost, u, bytes.NewReader(body))
	if err != nil {
		return err
	}
	req.GetBody = func() (io.ReadCloser, error) {
		return io.NopCloser(bytes.NewReader(body)), nil
	}
	req.Header.Set("Authorization", "Bearer "+strings.TrimSpace(deps.APIToken))
	req.Header.Set("Content-Type", "application/json")
	if idempotencyKey != "" {
		req.Header.Set("Idempotency-Key", idempotencyKey)
	}
	hc := deps.HTTP
	if hc == nil {
		hc = httpclient.New(httpclient.DefaultTimeout)
	}
	resp, err := httpclient.DoWithRetry(ctx, hc, req, 3)
	if err != nil {
		return err
	}
	defer func() { _ = resp.Body.Close() }()
	if resp.StatusCode == http.StatusNoContent || resp.StatusCode == http.StatusAccepted {
		return nil
	}
	b, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
	return fmt.Errorf("%s: %s", resp.Status, strings.TrimSpace(string(b)))
}