felhom.eu/hub/internal/wgsync/reconciler.go

package wgsync

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"time"

	"gitea.dooplex.hu/admin/felhom-hub/internal/store"
)

// syncer is the push seam — satisfied by *Client; tests inject a fake to count calls and
// capture payloads without SSH.
type syncer interface {
	Push(ctx context.Context, payload []byte) error
}

// Reconciler keeps the endpoint's WG peer list converged on the hub DB (the source of truth).
// DECLARATIVE: every push carries the FULL desired list — never deltas — so endpoint drift
// (a manually-added peer, a missed earlier push) is erased on the next push, and a failed push
// self-heals on the next tick. This full-list property is load-bearing (Scenario D drift
// repair) — do not "optimize" it into deltas.
type Reconciler struct {
	store    *store.Store
	sync     syncer
	trigger  chan struct{}
	interval time.Duration // tick period; tests shrink it
	logger   *log.Logger
}

// NewReconciler builds a Reconciler over the store and a syncer (the SSH client in production).
func NewReconciler(st *store.Store, sync syncer, logger *log.Logger) *Reconciler {
	if logger == nil {
		logger = log.Default()
	}
	return &Reconciler{
		store:    st,
		sync:     sync,
		trigger:  make(chan struct{}, 1),
		interval: 5 * time.Minute,
		logger:   logger,
	}
}

// Trigger requests an immediate sync from Run's loop. Non-blocking: a pending trigger already
// covers this request (the full list is pushed either way).
func (r *Reconciler) Trigger() {
	select {
	case r.trigger <- struct{}{}:
	default:
	}
}

// payload builds the versioned, deterministic sync document from the store.
// {"version":1,"interface":"wg0","peers":[{"pubkey":"...","allowed_ip":"<ip>/32"}]}
func (r *Reconciler) payload() ([]byte, error) {
	peers, err := r.store.ListWGPeers()
	if err != nil {
		return nil, fmt.Errorf("list peers: %w", err)
	}
	type wirePeer struct {
		Pubkey    string `json:"pubkey"`
		AllowedIP string `json:"allowed_ip"`
	}
	doc := struct {
		Version   int        `json:"version"`
		Interface string     `json:"interface"`
		Peers     []wirePeer `json:"peers"`
	}{Version: 1, Interface: "wg0", Peers: make([]wirePeer, 0, len(peers))}
	for _, p := range peers {
		doc.Peers = append(doc.Peers, wirePeer{Pubkey: p.Pubkey, AllowedIP: p.AssignedIP + "/32"})
	}
	return json.Marshal(doc)
}

// SyncNow builds the current full-list payload and pushes it. Called inline by the mutation
// handlers (short ctx) and by Run's loop.
func (r *Reconciler) SyncNow(ctx context.Context) error {
	payload, err := r.payload()
	if err != nil {
		return err
	}
	return r.sync.Push(ctx, payload)
}

// Run loops until ctx is done: an explicit Trigger OR the periodic tick pushes the full list.
// The periodic push happens even with zero mutations — that is the drift repair. Errors are
// logged and retried on the next signal; Run never exits early.
func (r *Reconciler) Run(ctx context.Context) {
	ticker := time.NewTicker(r.interval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-r.trigger:
		case <-ticker.C:
		}
		syncCtx, cancel := context.WithTimeout(ctx, 60*time.Second)
		if err := r.SyncNow(syncCtx); err != nil {
			r.logger.Printf("[ERROR] wgsync reconcile: %v (will retry on next tick)", err)
		}
		cancel()
	}
}