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/jobs"
	"evobgp/internal/pipeline"
	"evobgp/internal/store"
)

// Deps wires the scheduler to the store. Use either Jobs (same process as API, e.g. evobgp-all) or
// APIBase+APIToken to call the control plane over HTTP (separate containers in reference Compose).
type Deps struct {
	Store    store.Backend
	Jobs     *jobs.Registry
	APIBase  string // e.g. http://evobgp-api:8080
	APIToken string // Bearer token (operator/editor)
	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) == "") {
		log.Fatalf("evobgp-scheduler: need either in-process Jobs (evobgp-all) or both EVOBGP_CONTROL_PLANE_URL and EVOBGP_SCHEDULER_BEARER")
	}
	t := time.NewTicker(30 * time.Second)
	defer t.Stop()
	if deps.Jobs != nil {
		log.Printf("evobgp-scheduler: active (in-process enqueue tenant_refresh)")
	} else {
		log.Printf("evobgp-scheduler: active (HTTP POST .../tenant/refresh → %s)", strings.TrimSpace(deps.APIBase))
	}
	for {
		select {
		case <-ctx.Done():
			log.Printf("evobgp-scheduler: stopped")
			return
		case <-t.C:
			tick(context.Background(), deps)
		}
	}
}

func tick(ctx context.Context, deps *Deps) {
	tenants, err := deps.Store.ListTenantIDs()
	if err != nil {
		log.Printf("evobgp-scheduler: list tenants: %v", err)
		return
	}
	now := time.Now().UTC()
	tickBucket := now.Unix() / pipeline.SchedulerTickSec
	for _, tid := range tenants {
		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.Enqueue(tid, jobs.KindTenantRefresh, &key, nil, map[string]any{
				"module_ids": ids,
				"trigger":    "scheduler",
			})
			if err != nil {
				log.Printf("evobgp-scheduler: enqueue tenant %s: %v", tid, err)
				continue
			}
			if created {
				log.Printf("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 {
			log.Printf("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.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 = http.DefaultClient
	}
	resp, err := hc.Do(req)
	if err != nil {
		return err
	}
	defer 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)))
}