hypercache_dist.go

package hypercache

import (
	"context"

	"github.com/hyp3rd/hypercache/internal/eventbus"
	"github.com/hyp3rd/hypercache/pkg/backend"
)

// EventBus returns the in-process broadcaster the distributed
// backend uses to publish topology changes (`members`,
// `heartbeat`). Returns nil when the backend is not DistMemory —
// the management HTTP server's SSE handler treats nil as
// "streaming unsupported" and falls back to a 503.
//
// The bus is owned by DistMemory and lives until the backend's
// lifecycle context is cancelled. SSE handlers Subscribe via the
// request context so they're reaped when either the client
// disconnects or the cache shuts down.
func (hyperCache *HyperCache[T]) EventBus() *eventbus.Bus {
	if dm, ok := any(hyperCache.backend).(*backend.DistMemory); ok {
		return dm.EventBus()
	}

	return nil
}

// DistMetrics returns distributed backend metrics if the underlying backend is DistMemory.
// Returns nil if unsupported.
func (hyperCache *HyperCache[T]) DistMetrics() any { // generic any to avoid exporting type into core interface
	if dm, ok := any(hyperCache.backend).(*backend.DistMemory); ok {
		m := dm.Metrics()

		return m
	}

	return nil
}

// ClusterOwners returns the owners for a key if the distributed backend supports it; otherwise empty slice.
func (hyperCache *HyperCache[T]) ClusterOwners(key string) []string {
	if dm, ok := any(hyperCache.backend).(*backend.DistMemory); ok {
		owners := dm.DebugOwners(key)

		out := make([]string, 0, len(owners))
		for _, o := range owners {
			out = append(out, string(o))
		}

		return out
	}

	return nil
}

// ClusterKeys enumerates keys across the whole cluster, optionally
// filtered by `pattern` (prefix when no glob metacharacters are
// present, glob via path.Match otherwise — see backend.ListKeys for
// the canonical semantics). `maxResults` caps the merged-and-deduped
// result set; 0 picks the backend default. Returns a passthrough
// pointer-typed result, or nil when the backend isn't a DistMemory
// (callers treat nil as "feature unsupported" and surface 501/404).
//
// Best-effort: peer fan-out failures populate `PartialNodes` rather
// than failing the whole call, mirroring read-repair / hint-replay
// semantics elsewhere.
func (hyperCache *HyperCache[T]) ClusterKeys(
	ctx context.Context,
	pattern string,
	maxResults int,
) (*backend.ListKeysResult, error) {
	dm, ok := any(hyperCache.backend).(*backend.DistMemory)
	if !ok {
		return nil, nil //nolint:nilnil // explicit "feature unsupported" sentinel for callers to detect
	}

	res, err := dm.ListKeys(ctx, pattern, maxResults)
	if err != nil {
		return nil, err
	}

	return &res, nil
}

// DistMembershipSnapshot returns a snapshot of membership if distributed backend; otherwise nil slice.
//
//nolint:nonamedreturns
func (hyperCache *HyperCache[T]) DistMembershipSnapshot() (members []struct {
	ID          string
	Address     string
	State       string
	Incarnation uint64
}, replication, vnodes int,
) {
	if dm, ok := any(hyperCache.backend).(*backend.DistMemory); ok {
		membership := dm.Membership()
		ring := dm.Ring()

		if membership == nil || ring == nil {
			return nil, 0, 0
		}

		nodes := membership.List()
		out := make([]struct {
			ID          string
			Address     string
			State       string
			Incarnation uint64
		}, 0, len(nodes))

		for _, node := range nodes {
			out = append(out, struct {
				ID          string
				Address     string
				State       string
				Incarnation uint64
			}{
				ID:          string(node.ID),
				Address:     node.Address,
				State:       node.State.String(),
				Incarnation: node.Incarnation,
			})
		}

		return out, ring.Replication(), ring.VirtualNodesPerNode()
	}

	return nil, 0, 0
}

// DistRingHashSpots returns vnode hashes as hex strings if available (debug).
func (hyperCache *HyperCache[T]) DistRingHashSpots() []string {
	if dm, ok := any(hyperCache.backend).(*backend.DistMemory); ok {
		if ring := dm.Ring(); ring != nil {
			return ring.VNodeHashes()
		}
	}

	return nil
}

// DistDrain marks the underlying distributed backend for graceful
// shutdown when one is configured: /health flips to 503 on the dist
// HTTP listener, Set/Remove return sentinel.ErrDraining, Get
// continues to serve. Returns nil when the backend is not a
// DistMemory (no-op for in-memory and Redis backends), so callers
// don't need to type-check before invoking it.
//
// One-way and idempotent. Operators clear it by restarting the
// process after Drain settles and the cache has been Stopped.
func (hyperCache *HyperCache[T]) DistDrain(ctx context.Context) error {
	dm, ok := any(hyperCache.backend).(*backend.DistMemory)
	if !ok {
		return nil
	}

	return dm.Drain(ctx)
}

// DistHeartbeatMetrics returns distributed heartbeat metrics if supported.
func (hyperCache *HyperCache[T]) DistHeartbeatMetrics() any {
	if dm, ok := any(hyperCache.backend).(*backend.DistMemory); ok {
		m := dm.Metrics()

		return map[string]any{
			"heartbeatSuccess":   m.HeartbeatSuccess,
			"heartbeatFailure":   m.HeartbeatFailure,
			"nodesRemoved":       m.NodesRemoved,
			"readPrimaryPromote": m.ReadPrimaryPromote,
		}
	}

	return nil
}