forgeext

package
v0.0.6 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 16, 2026 License: Apache-2.0 Imports: 34 Imported by: 0

Documentation

Index

Constants

View Source 
const Version = "0.1.0"

Version is the fabriq forge extension version.

Variables

This section is empty.

Functions

func LoadConfig 

func LoadConfig(cm forge.ConfigManager, prefix string) fabriq.Config

LoadConfig builds a fabriq.Config from a forge ConfigManager. prefix is "" for the top-level key contract (cmd/fabriq serve) or "extensions.fabriq." for the first-class host-app convention. Relocated and parameterized from cmd/fabriq's loadFabriqConfig; the elasticsearch.addrs GetStringSlice handling is preserved (confy does not split a comma env string into a Go slice).

Types

type Config 

type Config struct {
	Fabriq            fabriq.Config
	RunWorker         bool
	ReconcileInterval time.Duration
	// DocCompactInterval is how often the document-plane loop runs the
	// SnapshotEvery compaction sweep (the sweep scans and aggregates the
	// whole update-log table, so it is deliberately much slower than the
	// per-second materializer). Zero falls back to 30s at run time.
	DocCompactInterval time.Duration
	// BlobGCGrace protects freshly-created CAS entries and orphan bytes from
	// collection for this window. Zero falls back to 1h at run time.
	BlobGCGrace time.Duration
	// RollupInterval is how often the leader-elected rollup:insights
	// maintainer runs one pass over every materialized metric (phase 2b).
	// Unlike ReconcileInterval (opt-in, disabled at zero), this job defaults
	// ON once Insights is enabled and at least one metric declares a Rollup:
	// zero falls back to 1 minute at run time, so materialized rollups stay
	// current without extra configuration.
	RollupInterval time.Duration
	// Embedder enables the embedding worker: each write to an entity with an
	// EmbedSpec is embedded + vector-upserted asynchronously. Nil = disabled.
	Embedder agent.Embedder
	// Summarizer enables the distillation worker: each write to an entity with
	// a DistillSpec is summarized into its digest tree asynchronously (debounced,
	// per-tenant single-flight). Nil = distillation disabled.
	Summarizer agent.Summarizer
	// Guard is the optional PII/guardrail seam for distillation (nil = identity).
	Guard agent.Guard
	// DistillFailOpenGuard flips the guard from fail-closed (default) to fail-open.
	DistillFailOpenGuard bool
	// DistillRecipeVersion salts the digest ContentHash; bump to rebuild the tree.
	DistillRecipeVersion string
	// DistillDebounce is the per-tenant coalescing window for L0+rollup sweeps.
	DistillDebounce time.Duration
	// DistillMaxWait caps how long a continuously-written tenant's sweep can be
	// deferred by debounce resets. Zero falls back to 10×DistillDebounce.
	// A value smaller than DistillDebounce is clamped up to DistillDebounce.
	DistillMaxWait time.Duration
	// Clusterer overrides the default in-core multi-probe SimHash clusterer.
	Clusterer agent.Clusterer
	// GroveDatabase names the *grove.DB to borrow from the host DI container
	// when no Postgres DSN/shards are configured. Empty resolves the default
	// (unnamed) grove.DB. Mirrors xraph/authsome's GroveDatabase knob. Read
	// from extensions.fabriq.groveDatabase when unset programmatically.
	GroveDatabase string
}

Config is the fabriq forge extension's configuration: the data-fabric config plus worker knobs. Build it with options; the extension overlays values from the config manager under extensions.fabriq.* at Register (options win).

type Extension 

type Extension struct {
	forge.BaseExtension
	// contains filtered or unexported fields
}

Extension exposes the fabriq data fabric as a first-class Forge extension: the facade as a DI service (alias "fabriq"), auto health, fabriq's migrations (Task 3), and an opt-in background worker (Task 4).

func New 

func New(reg *registry.Registry, opts ...Option) *Extension

New creates a new Extension with the given registry and options.

func (*Extension) Config added in v0.0.6 

func (e *Extension) Config() fabriq.Config

Config returns the resolved fabriq data-fabric config (datastore DSNs, catalog clusters, store addresses) after the Register-time overlay. The adminapi connection-info endpoints read it to surface REDACTED topology; it carries secrets, so callers MUST redact before serializing. Locked read: the overlay in Register may replace it before Start.

func (*Extension) Dependencies 

func (e *Extension) Dependencies() []string

func (*Extension) Description 

func (e *Extension) Description() string

func (*Extension) Fabriq 

func (e *Extension) Fabriq() *fabriq.Fabriq

Fabriq returns the opened facade (nil before Start).

func (*Extension) Health 

func (e *Extension) Health(ctx context.Context) error

Health implements forge.Extension. Pings the Postgres store.

func (*Extension) Migrate 

func (e *Extension) Migrate(ctx context.Context) (*forge.MigrationResult, error)

Migrate runs fabriq's pending migrations forward (forge MigratableExtension).

It opens a fresh grove Orchestrator against the primary DSN, runs all pending migrations, and translates the grove MigrateResult into a forge.MigrationResult.

func (*Extension) MigrationStatus 

func (e *Extension) MigrationStatus(ctx context.Context) ([]*forge.MigrationGroupInfo, error)

MigrationStatus returns the current state of fabriq's migrations grouped by migration group (forge MigratableExtension).

grove's Status method returns []*migrate.GroupStatus, each with Applied and Pending []*migrate.MigrationStatus slices. This is translated to the forge type hierarchy: one *forge.MigrationGroupInfo per grove GroupStatus, with *forge.MigrationInfo entries in Applied and Pending.

func (*Extension) Name 

func (e *Extension) Name() string

func (*Extension) Provisioner added in v0.0.6 

func (e *Extension) Provisioner() *provision.Provisioner

Provisioner builds a tenant provisioner over the ALREADY-OPEN catalog store (reused, not a second connection) and the configured clusters. Nil outside catalog mode — the admin API uses that to gate its tenant endpoints. Safe to call per request: New is cheap (no I/O).

func (*Extension) Register 

func (e *Extension) Register(app forge.App) error

Register implements forge.Extension. It MUST call e.BaseExtension.Register first.

func (*Extension) RemoteHandler added in v0.0.6 

func (e *Extension) RemoteHandler() *remote.Handler

RemoteHandler returns a remote.Handler that serves this extension's embedded facade over the fabriq remote protocol (ADR 0009). It is the grpc-free half of the server topology: a *grpc.Server is stood up over it by remotegrpc.NewServer in the remote/grpc module, keeping google.golang.org/grpc out of this module. Nil before Start (no facade yet), mirroring Fabriq().

func (*Extension) ResolveGrove added in v0.0.3 

func (e *Extension) ResolveGrove() *grove.DB

ResolveGrove best-effort resolves a *grove.DB from the host app's DI container (honouring GroveDatabase), returning nil when none is available. It is the exported seam wrapping extensions (e.g. a consuming app's fabriqkg) use to borrow the same handle fabriq serves and migrates against.

func (*Extension) Rollback 

func (e *Extension) Rollback(ctx context.Context) (*forge.MigrationResult, error)

Rollback rolls back the last applied fabriq migration (forge MigratableExtension).

grove's Rollback rolls back exactly one migration (the most recently applied in the group). RolledBack will be 0 or 1.

func (*Extension) Run 

func (e *Extension) Run(ctx context.Context) error

Run implements forge.RunnableExtension: supervise the leader-elected relay until shutdown. If RunWorker is false this is a no-op.

func (*Extension) SchemaMode added in v0.0.6 

func (e *Extension) SchemaMode() bool

SchemaMode reports whether catalog mode uses schema-per-tenant isolation.

func (*Extension) SchemaProvisioner added in v0.0.6 

func (e *Extension) SchemaProvisioner() *provision.SchemaProvisioner

SchemaProvisioner builds a schema-per-tenant provisioner over the already-open catalog store. Nil outside schema-isolation catalog mode — the admin API uses that to pick the provisioning path.

func (*Extension) Shutdown 

func (e *Extension) Shutdown(ctx context.Context) error

Shutdown implements forge.RunnableExtension: SIGTERM drain. If RunWorker is false (or Run was never called), this is a no-op.

func (*Extension) Start 

func (e *Extension) Start(ctx context.Context) error

Start implements forge.Extension. Opens the fabriq facade.

func (*Extension) Stop 

func (e *Extension) Stop(_ context.Context) error

Stop implements forge.Extension. Closes the fabriq facade.

func (*Extension) Stores 

func (e *Extension) Stores() *fabriq.Stores

Stores returns the opened adapters (nil before Start). The gateway extension reads Stores().Redis to build the live-query cluster transport.

func (*Extension) Version 

func (e *Extension) Version() string

type GatewayConfig 

type GatewayConfig struct {
	// GatewayID is this gateway's cluster identity (deltas are routed back to it
	// over lq:delta:{id}). Defaults to "gw-<host>-<pid>".
	GatewayID string
	// BasePath is the SSE endpoint path; the WebSocket endpoint is BasePath+"/ws".
	// Defaults to "/api/v1/live".
	BasePath string
	// HeartbeatTTL is the membership liveness window of the Redis cluster
	// transport (default 6s). Shards must refresh within it.
	HeartbeatTTL time.Duration
	// SSE and WS tune the per-connection delivery loops (heartbeat, write
	// deadlines/watchdog).
	SSE gateway.SSEOptions
	WS  gateway.WSOptions
	// RouteOptions are forwarded verbatim to router.SSE/router.WebSocket, so the
	// host app attaches its own auth middleware and AsyncAPI/OpenAPI schemas
	// (e.g. pkgAuth.RequirePermission(...), forge.WithSSEMessage(...)). fabriq
	// stays auth-scheme-agnostic.
	RouteOptions []forge.RouteOption
	// EnableDocEndpoints mounts the CRDT document endpoints (docs/update,
	// docs/sync, docs/subscribe, docs/presence) under BasePath. OPT-IN:
	// update is a write surface, so a live-query-only gateway must not
	// grow it silently on upgrade.
	EnableDocEndpoints bool
}

GatewayConfig configures the standalone live-query gateway tier: the edge that terminates client SSE/WebSocket connections and routes them over the sharded control/delta protocol to the matcher shards.

type GatewayExtension 

type GatewayExtension struct {
	forge.BaseExtension
	// contains filtered or unexported fields
}

GatewayExtension exposes the fabriq live-query gateway as a Forge extension: it builds a cluster.Gateway over the fabriq facade's Redis transport, runs its demux pump, and registers the SSE + WebSocket controllers. It depends on the "fabriq" extension (it reads its Stores().Redis), so it starts after it.

func NewGateway 

func NewGateway(fab *Extension, opts ...GatewayOption) *GatewayExtension

NewGateway builds the gateway extension wired to a started fabriq Extension.

func (*GatewayExtension) Dependencies 

func (g *GatewayExtension) Dependencies() []string

func (*GatewayExtension) Description 

func (g *GatewayExtension) Description() string

func (*GatewayExtension) Name 

func (g *GatewayExtension) Name() string

func (*GatewayExtension) Register 

func (g *GatewayExtension) Register(app forge.App) error

Register registers the SSE and WebSocket controllers. Their handlers resolve the backend lazily (it is built in Start), the same lazy-DI pattern the fabriq facade uses — safe because requests only arrive after Start.

func (*GatewayExtension) Start 

func (g *GatewayExtension) Start(_ context.Context) error

Start builds the cluster.Gateway over the facade's Redis transport and runs its demux pump.

func (*GatewayExtension) Stop 

func (g *GatewayExtension) Stop(ctx context.Context) error

Stop stops the gateway demux pump.

func (*GatewayExtension) Version 

func (g *GatewayExtension) Version() string

type GatewayOption 

type GatewayOption func(*GatewayConfig)

GatewayOption is a functional option for GatewayConfig.

func WithGatewayBasePath 

func WithGatewayBasePath(p string) GatewayOption

WithGatewayBasePath sets the SSE endpoint path (WS is BasePath+"/ws").

func WithGatewayDocEndpoints added in v0.0.5 

func WithGatewayDocEndpoints() GatewayOption

WithGatewayDocEndpoints mounts the CRDT document sync + presence endpoints (opt-in: docs/update is a write surface — pair it with fabriq.WithDocumentAuthz and auth middleware via route options).

func WithGatewayHeartbeatTTL 

func WithGatewayHeartbeatTTL(d time.Duration) GatewayOption

WithGatewayHeartbeatTTL sets the cluster membership liveness window.

func WithGatewayID 

func WithGatewayID(id string) GatewayOption

WithGatewayID sets the gateway's cluster identity.

func WithGatewayRouteOptions 

func WithGatewayRouteOptions(opts ...forge.RouteOption) GatewayOption

WithGatewayRouteOptions appends route options forwarded to the SSE/WS routes (auth, AsyncAPI/OpenAPI documentation).

func WithGatewaySSEHeartbeat 

func WithGatewaySSEHeartbeat(d time.Duration) GatewayOption

WithGatewaySSEHeartbeat sets the SSE keep-alive interval.

func WithGatewayWriteTimeout 

func WithGatewayWriteTimeout(d time.Duration) GatewayOption

WithGatewayWriteTimeout bounds a single SSE/WS write before the connection is torn down (the client reconnects to a fresh snapshot).

type Option 

type Option func(*Config)

Option is a functional option for Config.

func WithBlobGCGrace 

func WithBlobGCGrace(d time.Duration) Option

WithBlobGCGrace sets the grace window before an unreferenced CAS entry or orphan byte becomes GC-eligible. Defaults to 1h when zero.

func WithClusterer added in v0.0.3 

func WithClusterer(c agent.Clusterer) Option

WithClusterer overrides the default multi-probe SimHash clusterer (e.g. a vector-based gmmclusterer). Nil = default.

func WithConfig 

func WithConfig(c fabriq.Config) Option

WithConfig sets the underlying fabriq data-fabric configuration.

func WithCustomAppliers 

func WithCustomAppliers(a ...projection.CustomApplier) Option

WithCustomAppliers appends consumer-supplied projection appliers to the fabriq config. They are unioned after the built-in declarative applier for their Target and MUST be pure (see projection.CustomApplier).

func WithDistillDebounce 

func WithDistillDebounce(d time.Duration) Option

WithDistillDebounce sets the per-tenant coalescing window for L0+rollup sweeps.

func WithDistillFailOpenGuard 

func WithDistillFailOpenGuard(v bool) Option

WithDistillFailOpenGuard flips the guard from fail-closed (default) to fail-open.

func WithDistillMaxWait added in v0.0.3 

func WithDistillMaxWait(d time.Duration) Option

WithDistillMaxWait caps how long a continuously-written tenant's sweep can be deferred by debounce resets. Zero falls back to 10×DistillDebounce. A value smaller than DistillDebounce is clamped up to DistillDebounce.

func WithDistillRecipeVersion 

func WithDistillRecipeVersion(v string) Option

WithDistillRecipeVersion salts the digest ContentHash; bump to rebuild the tree.

func WithDocCompactInterval added in v0.0.6 

func WithDocCompactInterval(d time.Duration) Option

WithDocCompactInterval sets how often the document-plane loop runs the SnapshotEvery compaction sweep (default 30s; integration tests use a fast interval).

func WithEmbedder 

func WithEmbedder(e agent.Embedder) Option

WithEmbedder enables the embedding worker: each write to an entity with an EmbedSpec is embedded + vector-upserted asynchronously. Nil = disabled.

func WithGroveDatabase added in v0.0.3 

func WithGroveDatabase(name string) Option

WithGroveDatabase names the *grove.DB to borrow from the host DI container when no Postgres DSN/shards are configured (empty = the default unnamed grove.DB). Mirrors xraph/authsome's grove-database selection.

func WithGuard 

func WithGuard(g agent.Guard) Option

WithGuard sets the optional PII/guardrail seam for distillation.

func WithReconcileInterval 

func WithReconcileInterval(d time.Duration) Option

WithReconcileInterval sets the interval at which the background worker reconciles projection state.

func WithRollupInterval added in v0.0.6 

func WithRollupInterval(d time.Duration) Option

WithRollupInterval sets how often the rollup:insights maintainer runs a pass over every materialized metric. Zero falls back to 1 minute at run time (integration tests use a fast interval so the pass fires promptly).

func WithSummarizer 

func WithSummarizer(s agent.Summarizer) Option

WithSummarizer enables the distillation worker: each write to an entity with a DistillSpec is summarized into its digest tree asynchronously. Nil = disabled.

func WithWorker 

func WithWorker(on bool) Option

WithWorker enables or disables the background reconcile worker.

Source Files

View all Source files

Directories

Path Synopsis
Package adminapi exposes a read-only admin HTTP surface over the fabriq query facade.
 Package adminapi exposes a read-only admin HTTP surface over the fabriq query facade.
Package agentmcp exposes the fabriq agent toolkit over MCP (JSON-RPC 2.0).
 Package agentmcp exposes the fabriq agent toolkit over MCP (JSON-RPC 2.0).
Package gmmclusterer is an optional, host-wired vector clusterer for fabriq context distillation.
 Package gmmclusterer is an optional, host-wired vector clusterer for fabriq context distillation.
Package shieldguard adapts github.com/xraph/shield to the agent.Guard seam.
 Package shieldguard adapts github.com/xraph/shield to the agent.Guard seam.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.