remote

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: 20 Imported by: 0

Documentation

Overview

Package remote is the OPTIONAL server face of fabriq: it lets backend services talk to a central, connection-owning fabriq deployment over gRPC instead of embedding the library and owning their own datastore pools. It changes the DEPLOYMENT TOPOLOGY, not the engine — Fabric implements the same core/query.Fabric interface application code already holds, so the call sites are identical (ADR 0009).

Layering. The client (Fabric) and server (Handler) sit on a narrow, codec-neutral Transport seam so the request/response envelope can be exercised — and unit-tested via Loopback — before the gRPC+protobuf binding (proto/fabriq/v1/fabriq.proto) is generated and wired. The production Transport is gRPC over HTTP/2 with mTLS; tenant and principal travel in call metadata, authenticated at the server edge, never trusted from a client field.

Scope. The write plane (Exec/ExecBatch), the relational and projection read ports (Get/GetMany/List, Graph/Search/Vector), the Timeseries and Spatial ports, the live plane (Subscribe plus maintained LiveQuery with Reanchor over a bidirectional stream), the blob byte plane (Put/Get/Head/Delete/List/Copy and the presign bypass) and the Document plane are all wired end-to-end, with registry-typed payload decoding and the typed-error taxonomy across the wire. Interactive (multi-round-trip) transactions remain a deliberate non-goal; raw-SQL Query, blob multipart/range, and the optional document history sub-interfaces are the remaining follow-ons (their accessors return ErrNotImplemented).

Index

Constants

View Source 
const (
	MethodExec               = "fabriq.v1.Fabriq/Exec"
	MethodExecBatch          = "fabriq.v1.Fabriq/ExecBatch"
	MethodGet                = "fabriq.v1.Fabriq/Get"
	MethodGetMany            = "fabriq.v1.Fabriq/GetMany"
	MethodList               = "fabriq.v1.Fabriq/List"
	MethodSubscribe          = "fabriq.v1.Fabriq/Subscribe"
	MethodLiveQuery          = "fabriq.v1.Fabriq/LiveQuery"
	MethodPutBlob            = "fabriq.v1.Fabriq/PutBlob"
	MethodGetBlob            = "fabriq.v1.Fabriq/GetBlob"
	MethodHeadBlob           = "fabriq.v1.Fabriq/HeadBlob"
	MethodDeleteBlob         = "fabriq.v1.Fabriq/DeleteBlob"
	MethodPresignBlob        = "fabriq.v1.Fabriq/PresignBlob"
	MethodListBlob           = "fabriq.v1.Fabriq/ListBlob"
	MethodCopyBlob           = "fabriq.v1.Fabriq/CopyBlob"
	MethodVectorSimilar      = "fabriq.v1.Fabriq/VectorSimilar"
	MethodVectorUpsert       = "fabriq.v1.Fabriq/VectorUpsert"
	MethodVectorDelete       = "fabriq.v1.Fabriq/VectorDelete"
	MethodVectorDeleteByMeta = "fabriq.v1.Fabriq/VectorDeleteByMeta"
	MethodVectorGet          = "fabriq.v1.Fabriq/VectorGet"
	MethodSearch             = "fabriq.v1.Fabriq/Search"
	MethodGraphQuery         = "fabriq.v1.Fabriq/GraphQuery"
	MethodTSBulkWrite        = "fabriq.v1.Fabriq/TSBulkWrite"
	MethodTSRange            = "fabriq.v1.Fabriq/TSRange"
	MethodSpatialUpsert      = "fabriq.v1.Fabriq/SpatialUpsert"
	MethodSpatialWithin      = "fabriq.v1.Fabriq/SpatialWithin"
	MethodSpatialGet         = "fabriq.v1.Fabriq/SpatialGet"
	MethodSpatialDelete      = "fabriq.v1.Fabriq/SpatialDelete"
	MethodDocApplyUpdate     = "fabriq.v1.Fabriq/DocApplyUpdate"
	MethodDocSync            = "fabriq.v1.Fabriq/DocSync"
	MethodDocSnapshot        = "fabriq.v1.Fabriq/DocSnapshot"
	MethodDocCompact         = "fabriq.v1.Fabriq/DocCompact"
	// MethodBidiEcho is a diagnostic bidirectional method that echoes each frame
	// back prefixed with "echo:". It exists to exercise the BidiStream primitive
	// independently of the live-query plane.
	MethodBidiEcho = "fabriq.v1.Fabriq/BidiEcho"
)

Fully-qualified RPC method names, mirroring the proto service.

Variables

View Source 
var ErrNotImplemented = errors.New("remote: not implemented over the remote transport (see ADR 0009)")

ErrNotImplemented is returned by the methods the remote surface does not wire: raw-SQL Query, WaitForProjection, the projection-internal ApplyMutations / Graph.TraverseAndHydrate, and the optional blob (multipart/range) and document (history) sub-interfaces. The relational/projection reads, Timeseries, Spatial, blob List/Copy, the Document plane and live-query Reanchor are all wired now. It is deliberately distinct from ErrStoreNotConfigured: the store may well be configured server-side — the remote transport for that method just isn't built (ADR 0009).

Functions

This section is empty.

Types

type BidiStreamConn added in v0.0.6 

type BidiStreamConn interface {
	Send(frame []byte) error
	Recv() ([]byte, error) // io.EOF at clean end
	Close() error
}

BidiStreamConn is the client view of a bidirectional call: Send and Recv are independent and may interleave for the life of the stream. Recv returns io.EOF once the server-side handler returns cleanly. Close tears the stream down (the server observes ctx.Done).

type ClientStreamConn 

type ClientStreamConn interface {
	Send(frame []byte) error
	CloseAndRecv() (reply []byte, err error)
	Close() error
}

ClientStreamConn is the client view of a client-streaming call: send N frames, then CloseAndRecv for the single response.

type Fabric 

type Fabric struct {
	// contains filtered or unexported fields
}

Fabric is the client face: it implements core/query.Fabric by marshaling each call onto a Transport. Application code holds it exactly as it holds the embedded *fabriq.Fabriq — same interface, same call sites (ADR 0009). For typed repositories use query.For[T](reg, f.Relational()) rather than fabriq.For[T], which is bound to the concrete embedded facade.

func New 

func New(t Transport) *Fabric

New builds a Fabric over a Transport (gRPC in production, Loopback in tests).

func (*Fabric) Analytics added in v0.0.6 

func (r *Fabric) Analytics() query.AnalyticsQuerier

Analytics returns the client face of query.AnalyticsQuerier over the transport. Wiring the calls onto the transport is a later increment (ADR 0009 sequencing); until then every method returns ErrNotImplemented.

func (*Fabric) Blob 

func (r *Fabric) Blob() blob.Store

func (*Fabric) Document 

func (r *Fabric) Document() document.Store

Document returns the base document.Store wired over the transport (ApplyUpdate/Sync/Snapshot/Compact); the optional sub-interfaces (HistoryReader/SegmentLister/HistoryPurger) are a later increment. Blob streams bytes (Put/Get) and the presign bypass over the transport.

func (*Fabric) Exec 

func (r *Fabric) Exec(ctx context.Context, cmd command.Command) (command.Result, error)

Exec sends one command and reconstructs the result (or typed error).

func (*Fabric) ExecBatch 

func (r *Fabric) ExecBatch(ctx context.Context, cmds []command.Command) ([]command.Result, error)

ExecBatch sends N commands to run in one server-side transaction.

func (*Fabric) Graph 

func (r *Fabric) Graph() query.GraphQuerier

func (*Fabric) LiveQuery 

func (r *Fabric) LiveQuery(ctx context.Context, q livequery.LiveQuery) (livequery.Snapshot, <-chan livequery.LiveDelta, *LiveHandle, error)

LiveQuery registers a maintained-result-set subscription over the remote transport's bidirectional Live stream: it returns the initial ordered snapshot, a channel of enter/leave/move/update deltas, and a handle to Reanchor or tear it down. It mirrors *fabriq.Fabriq.LiveQuery. Close the handle — or cancel ctx — to end the subscription.

func (*Fabric) Relational 

func (r *Fabric) Relational() query.RelationalQuerier

func (*Fabric) Search 

func (r *Fabric) Search() query.SearchQuerier

func (*Fabric) Spatial 

func (r *Fabric) Spatial() query.SpatialQuerier

func (*Fabric) Subscribe 

func (r *Fabric) Subscribe(ctx context.Context, scope query.SubscribeScope) (<-chan query.Delta, error)

Subscribe opens the conflated channel-delta stream. The first frame is a handshake: a setup error (authz / scope resolution) returns synchronously, mirroring the in-process contract; otherwise a goroutine drains delta frames into the returned channel until the stream ends or ctx is cancelled.

func (*Fabric) Timeseries 

func (r *Fabric) Timeseries() query.TSQuerier

func (*Fabric) Vector 

func (r *Fabric) Vector() query.VectorQuerier

func (*Fabric) WaitForProjection 

func (r *Fabric) WaitForProjection(_ context.Context, _, _, _ string, _ int64) error

WaitForProjection is a read-your-writes helper; it rides the read plane.

type Handler 

type Handler struct {
	// contains filtered or unexported fields
}

Handler is the server face: it terminates the remote envelope and delegates to a real, embedded query.Fabric (an Open()ed *fabriq.Fabriq on the connection-owning tier). The gRPC service implementation calls Dispatch / DispatchStream (or the per-method handlers) from its generated stubs.

SECURITY: ctx MUST already carry the tenant and principal resolved from the call's transport metadata by an auth interceptor — never from a field in the decoded message. The embedded facade then enforces RLS and authz exactly as it does in-process (ADR 0009 §Security).

func NewHandler 

func NewHandler(fab query.Fabric, reg *registry.Registry) *Handler

NewHandler builds a Handler over the embedded facade and its registry (the registry is the schema authority used to decode opaque payloads). If the facade also implements LiveQuerier (the concrete *fabriq.Fabriq does), the remote LiveQuery plane is enabled; otherwise it returns ErrNotImplemented.

func (*Handler) BidiEcho added in v0.0.6 

func (h *Handler) BidiEcho(_ context.Context, recv func() ([]byte, error), send func([]byte) error) error

BidiEcho is a diagnostic bidirectional handler: it echoes each received frame back to the client prefixed with "echo:", until the client stops sending (io.EOF) or the call is cancelled. It exists to exercise the bidi transport primitive on its own — the live-query plane is the real consumer.

func (*Handler) CopyBlob added in v0.0.6 

func (h *Handler) CopyBlob(ctx context.Context, in []byte) ([]byte, error)

func (*Handler) DeleteBlob 

func (h *Handler) DeleteBlob(ctx context.Context, in []byte) ([]byte, error)

func (*Handler) Dispatch 

func (h *Handler) Dispatch(ctx context.Context, method string, in []byte) ([]byte, error)

Dispatch routes a unary call by its method name — the server-side mirror of Transport.Unary, used by Loopback and by a thin gRPC unary shim.

func (*Handler) DispatchBidi added in v0.0.6 

func (h *Handler) DispatchBidi(ctx context.Context, method string, recv func() ([]byte, error), send func([]byte) error) error

DispatchBidi routes a bidirectional call by method name. recv returns the next client frame (io.EOF when the client is done sending); send delivers one frame to the client. The handler returns when the call is over; the transport binding turns a returned error into the stream status.

func (*Handler) DispatchClientStream 

func (h *Handler) DispatchClientStream(ctx context.Context, method string, recv func() ([]byte, error)) ([]byte, error)

DispatchClientStream routes a client-streaming call by method name. recv returns the next request frame, or io.EOF when the client is done; the handler returns the single response frame.

func (*Handler) DispatchStream 

func (h *Handler) DispatchStream(ctx context.Context, method string, in []byte, send func([]byte) error) error

DispatchStream routes a server-streaming call by method name. send delivers one frame; the error it returns (e.g. client gone) aborts the stream.

func (*Handler) DocApplyUpdate added in v0.0.6 

func (h *Handler) DocApplyUpdate(ctx context.Context, in []byte) ([]byte, error)

func (*Handler) DocCompact added in v0.0.6 

func (h *Handler) DocCompact(ctx context.Context, in []byte) ([]byte, error)

func (*Handler) DocSnapshot added in v0.0.6 

func (h *Handler) DocSnapshot(ctx context.Context, in []byte) ([]byte, error)

func (*Handler) DocSync added in v0.0.6 

func (h *Handler) DocSync(ctx context.Context, in []byte) ([]byte, error)

func (*Handler) Exec 

func (h *Handler) Exec(ctx context.Context, in []byte) ([]byte, error)

Exec decodes one command, rebuilds its registry-typed payload, runs it on the embedded facade, and encodes the result or typed error.

func (*Handler) ExecBatch 

func (h *Handler) ExecBatch(ctx context.Context, in []byte) ([]byte, error)

ExecBatch decodes N commands and runs them in one server-side transaction.

func (*Handler) Get 

func (h *Handler) Get(ctx context.Context, in []byte) ([]byte, error)

Get is the server side of MethodGet: build a registry-typed scan target, run the real relational read, and return the row as opaque JSON.

func (*Handler) GetBlob 

func (h *Handler) GetBlob(ctx context.Context, in []byte, send func([]byte) error) error

GetBlob is the server side of MethodGetBlob: the first frame carries the object metadata (or a setup error), then one data frame per chunk.

func (*Handler) GetMany 

func (h *Handler) GetMany(ctx context.Context, in []byte) ([]byte, error)

GetMany is the server side of MethodGetMany: the batched (no-N+1) read.

func (*Handler) GraphQuery 

func (h *Handler) GraphQuery(ctx context.Context, in []byte) ([]byte, error)

func (*Handler) HeadBlob 

func (h *Handler) HeadBlob(ctx context.Context, in []byte) ([]byte, error)

HeadBlob, DeleteBlob and PresignBlob are unary.

func (*Handler) List 

func (h *Handler) List(ctx context.Context, in []byte) ([]byte, error)

List is the server side of MethodList: decode the structured filter — preferring the typed `structured` field (full numeric fidelity) and falling back to the legacy opaque-JSON `query` body when structured is unset (back-compat with pre-Task-C clients for one release) — run the real paged read into a registry-typed slice target, and return opaque-JSON rows.

func (*Handler) ListBlob added in v0.0.6 

func (h *Handler) ListBlob(ctx context.Context, in []byte) ([]byte, error)

ListBlob and CopyBlob are unary.

func (*Handler) LiveQuery 

func (h *Handler) LiveQuery(ctx context.Context, recv func() ([]byte, error), send func([]byte) error) error

LiveQuery is the server side of MethodLiveQuery, a bidirectional stream. The client's first frame carries the query; it opens a maintained-window subscription on the facade, sends the snapshot as the first server frame, then drains deltas to the client in a goroutine. Meanwhile it loops on recv for control frames: a Reanchor frame slides the window (in-process Handle.Reanchor) and returns the fresh snapshot as a server frame. It exits — Closing the engine handle — when the client stops sending (io.EOF), disconnects (ctx.Done), or the delta channel closes. A facade without LiveQuery answers ErrNotImplemented.

SendMsg is not safe for concurrent use, so a mutex serializes the delta pump and the reanchor replies onto the single send func.

func (*Handler) PresignBlob 

func (h *Handler) PresignBlob(ctx context.Context, in []byte) ([]byte, error)

func (*Handler) PutBlob 

func (h *Handler) PutBlob(ctx context.Context, recv func() ([]byte, error)) ([]byte, error)

PutBlob is the server side of MethodPutBlob: the first frame carries metadata, the rest carry bytes, which it pipes into the byte store's streaming Put.

func (*Handler) Search 

func (h *Handler) Search(ctx context.Context, in []byte) ([]byte, error)

func (*Handler) SpatialDelete added in v0.0.6 

func (h *Handler) SpatialDelete(ctx context.Context, in []byte) ([]byte, error)

func (*Handler) SpatialGet added in v0.0.6 

func (h *Handler) SpatialGet(ctx context.Context, in []byte) ([]byte, error)

func (*Handler) SpatialUpsert added in v0.0.6 

func (h *Handler) SpatialUpsert(ctx context.Context, in []byte) ([]byte, error)

func (*Handler) SpatialWithin added in v0.0.6 

func (h *Handler) SpatialWithin(ctx context.Context, in []byte) ([]byte, error)

func (*Handler) Subscribe 

func (h *Handler) Subscribe(ctx context.Context, in []byte, send func([]byte) error) error

Subscribe is the server side of MethodSubscribe. The embedded facade resolves the scope (authz + channel resolution happen there); the first frame is a handshake reporting setup success or a typed error so the client can honor Subscribe's synchronous-error contract, then one Delta per frame follows.

func (*Handler) TSBulkWrite added in v0.0.6 

func (h *Handler) TSBulkWrite(ctx context.Context, in []byte) ([]byte, error)

func (*Handler) TSRange added in v0.0.6 

func (h *Handler) TSRange(ctx context.Context, in []byte) ([]byte, error)

func (*Handler) VectorDelete 

func (h *Handler) VectorDelete(ctx context.Context, in []byte) ([]byte, error)

func (*Handler) VectorDeleteByMeta added in v0.0.3 

func (h *Handler) VectorDeleteByMeta(ctx context.Context, in []byte) ([]byte, error)

func (*Handler) VectorGet added in v0.0.3 

func (h *Handler) VectorGet(ctx context.Context, in []byte) ([]byte, error)

func (*Handler) VectorSimilar 

func (h *Handler) VectorSimilar(ctx context.Context, in []byte) ([]byte, error)

func (*Handler) VectorUpsert 

func (h *Handler) VectorUpsert(ctx context.Context, in []byte) ([]byte, error)

type LiveHandle 

type LiveHandle struct {
	// contains filtered or unexported fields
}

LiveHandle controls a remote maintained subscription over the bidirectional LiveQuery stream. It owns the single stream reader (deltas are demuxed to the caller's channel, reanchor snapshots to Reanchor's waiter) so Send and Recv never race across goroutines.

func (*LiveHandle) Close 

func (h *LiveHandle) Close()

Close tears the remote subscription down: it closes the bidi stream, which the server observes (ctx.Done) and uses to Close the underlying engine handle.

func (*LiveHandle) Reanchor 

func (h *LiveHandle) Reanchor(ctx context.Context, cursor *livequery.Cursor, limit int) (livequery.Snapshot, error)

Reanchor slides the maintained window to a new cursor anchor (and optionally a new size) mid-stream: it sends a Reanchor control frame on the bidi stream and blocks for the fresh snapshot the server returns, while deltas keep flowing on the delta channel. Concurrent Reanchor calls are serialized.

type LiveQuerier 

type LiveQuerier interface {
	LiveQuery(ctx context.Context, q livequery.LiveQuery) (livequery.Snapshot, <-chan livequery.LiveDelta, LiveSubscription, error)
}

LiveQuerier is the maintained-result-set surface the remote Live plane needs. It is NOT part of query.Fabric — LiveQuery lives on the concrete *fabriq.Fabriq — so the Handler adapts it from the facade (NewHandler); a facade without it makes the remote LiveQuery return ErrNotImplemented. The returned subscription carries Reanchor across the bidi stream.

type LiveSubscription added in v0.0.6 

type LiveSubscription interface {
	Reanchor(ctx context.Context, cursor *livequery.Cursor, limit int) (livequery.Snapshot, error)
	Close()
}

LiveSubscription is the control surface the server drives on a live subscription: tear it down (Close) or slide its window to a new anchor (Reanchor). The in-process *livequery.Handle satisfies it by construction, so a facade whose LiveQuery returns *livequery.Handle plugs in via the adapter in NewHandler; a test double can satisfy it directly.

type Loopback 

type Loopback struct{ Handler *Handler }

Loopback is an in-process Transport that dispatches straight to a Handler — no network, only the envelope (de)serialization itself. It exists so the client, the envelope and the server-side handlers can be round-tripped in a unit test before the gRPC binding exists.

func (Loopback) BidiStream added in v0.0.6 

func (l Loopback) BidiStream(ctx context.Context, method string) (BidiStreamConn, error)

BidiStream implements Transport: it runs the bidirectional handler in a goroutine wired to two channels — one the caller Sends into (drained by the handler's recv), one the handler sends into (drained by the caller's Recv). It mirrors the ClientStream/ServerStream loopbacks: ctx is cancelled once the handler returns so a pending Send/Recv unblocks instead of leaking, and Recv reports io.EOF (or the handler's error) once the outbound channel closes.

func (Loopback) ClientStream 

func (l Loopback) ClientStream(ctx context.Context, method string) (ClientStreamConn, error)

ClientStream implements Transport: it runs the client-streaming handler in a goroutine that pulls frames from a channel the caller Sends to and produces a single reply. The ctx is cancelled once the handler returns so a pending Send unblocks instead of leaking.

func (Loopback) ServerStream 

func (l Loopback) ServerStream(ctx context.Context, method string, in []byte) (Stream, error)

ServerStream implements Transport: it runs the streaming handler in a goroutine that pushes frames into a buffered channel the returned Stream drains. The ctx is made cancellable so Close (or client disconnect) stops the server goroutine instead of leaking it — every send is guarded by ctx.Done.

func (Loopback) Unary 

func (l Loopback) Unary(ctx context.Context, method string, in []byte) ([]byte, error)

Unary implements Transport.

type Stream 

type Stream interface {
	// Recv returns the next frame, (nil, io.EOF) at a clean end, or (nil, err).
	Recv() ([]byte, error)
	// Close releases the stream and signals the server to stop producing.
	Close() error
}

Stream is the client view of a server-streaming response.

type Transport 

type Transport interface {
	// Unary invokes a request/response method.
	Unary(ctx context.Context, method string, in []byte) (out []byte, err error)
	// ServerStream opens a server-streaming method: one request, a stream of
	// framed responses. Drain Recv until io.EOF (clean end) or a non-EOF error,
	// then Close to release it.
	ServerStream(ctx context.Context, method string, in []byte) (Stream, error)
	// ClientStream opens a client-streaming method: the client Sends N frames
	// then CloseAndRecv for the single response. Used by chunked blob upload.
	ClientStream(ctx context.Context, method string) (ClientStreamConn, error)
	// BidiStream opens a bidirectional method: the client and server both Send
	// and Recv frames independently for the lifetime of the call. Recv returns
	// io.EOF at a clean server-side end. Used by the live-query plane, where a
	// Reanchor control frame is sent mid-stream while deltas keep flowing.
	BidiStream(ctx context.Context, method string) (BidiStreamConn, error)
}

Transport is the codec- and wire-neutral seam the client and server halves sit on. The production binding is gRPC over HTTP/2 with protobuf framing and mTLS (see proto/fabriq/v1/fabriq.proto and ADR 0009); Loopback is the in-process binding used to exercise the envelope without a network.

The ctx passed here carries the tenant/principal; the gRPC Transport turns those into call metadata that the server edge authenticates. in/out are the marshaled envelope bytes — canonical JSON in this skeleton, protobuf once the stubs are generated; the planes above this seam do not care which.

Source Files

View all Source files

Directories

Path Synopsis

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.