postgres

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

Documentation

Overview

Package postgres is fabriq's Postgres adapter, built on grove's pg driver. It implements the relational, timeseries, vector and command store ports against the source of truth.

Tenancy is enforced in layers:

  1. Structurally: every operation runs inside a transaction stamped with SET LOCAL app.tenant_id (set_config(..., true)), and every generated query carries an explicit tenant predicate where applicable.
  2. In the database: RLS policies (FORCE) key on that setting, so even raw SQL through the escape hatch cannot cross tenants.
  3. Backstop: a grove pre-query/pre-mutation hook observes every query on both paths (grove >= a01144a fires hooks inside transactions too). It allows the stamped transaction path — RLS is the guard there — and DENIES any pool-path access to a tenant table, which in this architecture is always a bug, returning ErrTenantHookTripped and counting the trip. See docs/decisions/0002-tenancy-layers.md.

Dynamic-entity reads (IsDynamic() == true) scan rows into *[]map[string]any rather than typed structs. Grove's schema-aware Scan only handles struct/slice-of-struct destinations, so dynamic reads use the PgTx.QueryRows path: a driver.Rows cursor iterated manually with Columns()+Scan into individual any destinations, then assembled into maps. Static reads are byte-unchanged.

Index

Constants

View Source 
const (
	// LockKeyRelay guards the outbox relay.
	LockKeyRelay = int64(1001)
	// LockKeyReconciler guards the projection drift reconciler.
	LockKeyReconciler = int64(1002)
	// LockKeyDocumentPlane guards the document materializer + compactor.
	LockKeyDocumentPlane = int64(1003)
	// LockKeyBlobGC guards the blob CAS garbage collector.
	LockKeyBlobGC = int64(1004)
	// LockKeyRollup guards the rollup:insights materialized-rollup maintainer
	// (phase 2b).
	LockKeyRollup = int64(1005)
)

Advisory lock keys per singleton worker role, per database. Stable across versions; never reuse a key for a different role. In the static plane the forgeext worker holds these for the process lifetime (leader election); in catalog mode the sweeper try-claims them per pass — the SAME keys, so the two modes can never both work one database.

Variables

View Source 
var RawQueryTimeout = 15 * time.Second

RawQueryTimeout bounds a single QueryDynamicReadOnly call via statement_timeout. Exported so tests can shrink it; production uses the 15s default.

Functions

func LockKeyBlobGCFor added in v0.0.6 

func LockKeyBlobGCFor(schema string) int64

LockKeyBlobGCFor is the schema-scoped blob-GC lock key.

func LockKeyDocumentPlaneFor added in v0.0.6 

func LockKeyDocumentPlaneFor(schema string) int64

LockKeyDocumentPlaneFor is the schema-scoped document-plane lock key.

func LockKeyReconcilerFor added in v0.0.6 

func LockKeyReconcilerFor(schema string) int64

LockKeyReconcilerFor is the schema-scoped reconciler lock key.

func LockKeyRelayFor added in v0.0.6 

func LockKeyRelayFor(schema string) int64

LockKeyRelayFor is the schema-scoped relay lock key.

func PingDSN added in v0.0.6 

func PingDSN(ctx context.Context, dsn string) error

PingDSN opens a short-lived connection to dsn and runs SELECT 1, reporting reachability. It is the reachability probe behind the adminapi connection-info endpoints (per-cluster and per-tenant DBs, which have no persistent adapter). Bound it with a context deadline; grove dials lazily, so the query is the dial.

Types

type Adapter 

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

Adapter implements command.Store, query.RelationalQuerier, query.TSQuerier and query.VectorQuerier on grove/pgdriver.

func Open 

func Open(ctx context.Context, dsn string, reg *registry.Registry, opts ...Option) (*Adapter, error)

Open connects to Postgres and wires the tenant backstop.

func OpenWithGrove added in v0.0.3 

func OpenWithGrove(gdb *grove.DB, reg *registry.Registry, opts ...Option) (*Adapter, error)

OpenWithGrove builds an Adapter on top of a grove.DB that the caller already dialed — the seam the forge extension uses to back fabriq's source of truth with a *grove.DB resolved from the host's DI container (mirroring how xraph/authsome borrows the shared grove) instead of dialing its own DSN.

The grove MUST be backed by the pg driver (grove + pgdriver): fabriq's command/relational/timeseries/vector plane is Postgres-specific. The handle is BORROWED — Close is a no-op for it, leaving the host to own the connection lifecycle. The tenant backstop is still registered on the shared hook engine, so pool-path access to fabriq's tenant tables is denied exactly as on a self-dialed adapter (host queries to other tables pass through).

func (*Adapter) AddGuardedTable added in v0.0.4 

func (a *Adapter) AddGuardedTable(table string)

AddGuardedTable registers a runtime-added tenant table with the pool-path backstop. Call after EnsureDynamic for a runtime-defined dynamic entity.

func (*Adapter) AdvanceRollupWatermark added in v0.0.6 

func (a *Adapter) AdvanceRollupWatermark(ctx context.Context, metric string, bucket time.Time) error

AdvanceRollupWatermark upserts the (tenant, metric) rollup watermark to bucket, taking the GREATEST of the existing and new values so a concurrent or out-of-order call can never move the watermark backwards. scope_id is stored NULL (via NULLIF($2,”)) — the watermark is per (tenant, metric), not per scope (rollup_state's PK is (tenant_id, metric)); scope_id on this table is incidental, carried only for the same scope-aware-RLS shape the sibling insights tables share.

func (*Adapter) AggregateVersions 

func (a *Adapter) AggregateVersions(ctx context.Context, tenantID, entity string) (map[string]int64, error)

AggregateVersions reads id -> version for one entity of a tenant — the reconciler's truth side (projection.TruthVersions).

func (*Adapter) BackstopTrips 

func (a *Adapter) BackstopTrips() int64

BackstopTrips reports how many times the tenant backstop fired.

func (*Adapter) BulkWrite 

func (a *Adapter) BulkWrite(ctx context.Context, series string, points []query.Point) error

BulkWrite implements query.TSQuerier — the event-bypass telemetry path. One multi-row INSERT per call; no per-row outbox events (the worker publishes conflated deltas instead).

func (*Adapter) Close 

func (a *Adapter) Close() error

Close releases the connection pool. For a borrowed grove (OpenWithGrove) it is a no-op: the host owns the handle's lifecycle.

func (*Adapter) Delete 

func (a *Adapter) Delete(ctx context.Context, entity, id string) error

Delete implements query.VectorQuerier.

func (*Adapter) DeleteByMeta added in v0.0.3 

func (a *Adapter) DeleteByMeta(ctx context.Context, entity string, filter map[string]string) error

DeleteByMeta removes embeddings for (tenant, entity) matching the meta filter.

func (*Adapter) Documents 

func (a *Adapter) Documents() *DocStore

Documents returns the document-plane store.

func (*Adapter) Driver 

func (a *Adapter) Driver() *pgdriver.PgDB

Driver exposes the typed pg driver for worker-plane components living in this module (relay, leader, migrations CLI). Never hand it to application code.

func (*Adapter) DropDynamic added in v0.0.4 

func (a *Adapter) DropDynamic(ctx context.Context, table string) error

DropDynamic drops a dynamic entity's table (its indexes and RLS policies drop with it). Idempotent (IF EXISTS). Destructive — callers gate confirmation.

func (*Adapter) DropDynamicColumn added in v0.0.4 

func (a *Adapter) DropDynamicColumn(ctx context.Context, table, column string) error

DropDynamicColumn drops a domain column from a dynamic table. Structural columns are refused. Runs as schema owner. Idempotent (IF EXISTS).

func (*Adapter) EnsureDynamic 

func (a *Adapter) EnsureDynamic(ctx context.Context, ent *registry.Entity) error

EnsureDynamic creates or ADDITIVELY EVOLVES the Postgres table for a dynamic entity from its descriptor: structural columns (id, tenant_id, version), declared domain columns, a tenant index, any descriptor-declared secondary indexes, and tenant-isolation RLS — mirroring the patterns in migrations/0003_site_asset_tag.go and migrations/0004_rls_policies.go.

Additive-evolution policy

When called on an already-existing table (e.g. because the consumer changed the descriptor by adding columns or indexes), EnsureDynamic reconciles the schema ADDITIVELY:

  • Each domain column is emitted as "ALTER TABLE … ADD COLUMN IF NOT EXISTS …", so new columns appear and existing ones are left untouched.
  • Each index is emitted as "CREATE INDEX IF NOT EXISTS …", so new indexes are created and existing ones are left untouched.

DROPS, RENAMES, and TYPE CHANGES are NOT auto-applied. If a column or index is removed from the descriptor, the physical column/index remains — evolution is strictly additive. Consumers that need destructive changes must write an explicit migration.

Attempting to add a NOT-NULL column (without a DEFAULT) to a table that already has rows will fail at the Postgres level. This is intentional: the consumer must either supply a Default expression in the descriptor or add the column as nullable. fabriq does not work around this safety boundary.

This is the FENCED managed-DDL lane: fabriq manages DDL ONLY for dynamic entities. Static entities keep migrations as the authority; calling EnsureDynamic for a non-dynamic entity (Spec.Schema == nil) is an error.

Run as the schema owner (superuser DSN), not the RLS-constrained app role.

The entity must be registered in the registry BEFORE the adapter is constructed (fabriq.Open), so the pool-path tenant backstop includes the dynamic table in its guarded set; EnsureDynamic only creates the physical table, it does not register the entity.

func (*Adapter) EnsureRollupTable added in v0.0.6 

func (a *Adapter) EnsureRollupTable(ctx context.Context, m *registry.MetricSpec) error

EnsureRollupTable creates (idempotently) the per-metric materialized- rollup table for m, via the same owner/DDL exec seam EnsureDynamic uses (a.execDDL — the schema-owner pool path, NOT a tenant-scoped transaction: DDL is tenant-agnostic; the table's RLS policy enforces per-row tenant isolation once rows are written). Safe to call repeatedly (CREATE TABLE IF NOT EXISTS + idempotent RLS statements) AND cheap to call repeatedly: forgeext/rollup.go's rollupOne calls this on EVERY maintainer tick, for every (tenant, metric) pair, not just once at boot.

The RLS statements specifically are NOT simply re-run every time, even though rollupTableDDL/rollupRLSStatements build them as an unconditional `ALTER ... ENABLE/FORCE ROW LEVEL SECURITY; DROP POLICY IF EXISTS tenant_isolation; CREATE POLICY tenant_isolation ...` sequence: execDDL runs each statement in its OWN autocommit transaction (there is no single transaction spanning DROP+CREATE), so on a table with FORCE ROW LEVEL SECURITY, the instant the DROP commits and before the CREATE commits, the table has NO policy at all — a concurrent app-role query in that window hits Postgres's default-deny-under-FORCE-RLS behavior and gets an empty result, not an error. Doing that on every maintainer tick, forever, is a recurring (if narrow) empty-result window for live readers. So this probes pg_policies FIRST (rollupPolicyExists) and only runs the RLS statements when the policy does not already exist — the first EnsureRollupTable call for a given table still creates it (with RLS applied) as before; every subsequent call, including every later maintainer tick, is then a true no-op for the RLS statements (CREATE TABLE IF NOT EXISTS and the unique index were already idempotent no-ops on their own). RLS itself is never removed or weakened by this — only the redundant re-application is skipped.

When m declares a sketch measure (count_distinct/percentile), the rollup table needs timescaledb_toolkit's hyperloglog/tdigest types. This is a fail-fast boot check, not silent degradation: if the toolkit is not available in the target database, EnsureRollupTable returns a loud error naming the metric BEFORE attempting to create the table, rather than letting a bare "type hyperloglog does not exist" driver error surface. When toolkit IS available, `CREATE EXTENSION IF NOT EXISTS timescaledb_toolkit` runs first (idempotent) so the toolkit's types exist before the CREATE TABLE that references them. Additive-only rollup metrics never touch the toolkit at all.

func (*Adapter) ExecRawDDL added in v0.0.4 

func (a *Adapter) ExecRawDDL(ctx context.Context, stmt string) error

ExecRawDDL runs a single raw DDL statement as the adapter's role (the schema owner on single-role deployments). This is the ad-hoc DDL escape hatch: it is deliberately unfenced beyond the caller's own gating and single-statement validation. Callers MUST gate and validate before calling.

func (*Adapter) Get 

func (a *Adapter) Get(ctx context.Context, entity, id string, into any) error

Get implements query.RelationalQuerier.

func (*Adapter) GetMany 

func (a *Adapter) GetMany(ctx context.Context, entity string, ids []string, into any) error

GetMany implements the batched hydration contract: ONE query, results in ids order, missing rows skipped.

func (*Adapter) Grove 

func (a *Adapter) Grove() *grove.DB

func (*Adapter) InTenantTx 

func (a *Adapter) InTenantTx(ctx context.Context, fn func(ctx context.Context, tx command.Tx) error) error

InTenantTx implements command.Store: one Postgres transaction, tenant stamped via SET LOCAL, row writes and outbox appends inside it.

func (*Adapter) List 

func (a *Adapter) List(ctx context.Context, entity string, q query.ListQuery, into any) error

List implements equality-filtered, ordered paging. Filter and order columns are validated against the binding — unknown columns are rejected, which is also the SQL-injection guard.

func (*Adapter) MaintainRollup added in v0.0.6 

func (a *Adapter) MaintainRollup(ctx context.Context, m *registry.MetricSpec, now time.Time) error

MaintainRollup runs one maintainer pass for m: seals buckets whose end has passed SealGrace, aggregates the sealable range (plus a trailing RerollWindow re-roll to absorb late arrivals into already-sealed buckets), and advances the watermark. A no-op (returns nil without touching the rollup table or watermark) when nothing new has sealed since the last pass.

  • sealBefore = now - SealGrace (SealGrace defaults to defaultSealGrace when the metric leaves it at zero).
  • to = sealBefore truncated to the metric's Bucket — the newest bucket boundary that is fully sealed.
  • from: on a metric with an existing watermark, `wm - RerollWindow` (RerollWindow defaults to defaultRerollMultiple*Bucket when left at zero) — re-rolling the trailing window absorbs events that arrived late (within grace) but after an earlier pass already sealed that bucket. On a metric's first pass (no watermark row yet), from is the zero time.Time — a far-past floor — so the first pass rolls up every historical sealed bucket.
  • Guard: if to does not come strictly after from, nothing has newly sealed since the last pass (or ever, on a fresh metric with no data yet) — no-op, watermark untouched.

func (*Adapter) NewLiveStore 

func (a *Adapter) NewLiveStore() *LiveStore

NewLiveStore returns the live query snapshot/refill store for this adapter.

func (*Adapter) Ping added in v0.0.6 

func (a *Adapter) Ping(ctx context.Context) error

Grove exposes the grove handle (hook-guarded pool path) for advanced embedding. Tenant tables are NOT reachable through it — the backstop denies them; use the fabric ports. Ping verifies the database actually accepts connections — grove/pgx dial lazily, so Open succeeding does not prove the database exists. The catalog-mode dialer pings before handing the shard to the pool, so a dead tenant database fails AT DIAL TIME and the pool's breaker opens instead of every query failing individually.

func (*Adapter) ProjectionState 

func (a *Adapter) ProjectionState() *StateRepo

ProjectionState returns the projection bookkeeping repo.

func (*Adapter) Query 

func (a *Adapter) Query(ctx context.Context, into any, sql string, args ...any) error

Query is the raw SQL escape hatch for reads. It still runs inside a tenant-stamped transaction, so RLS contains it; tables outside RLS (guarded tables) additionally require a literal tenant_id reference.

func (*Adapter) QueryDynamicReadOnly added in v0.0.4 

func (a *Adapter) QueryDynamicReadOnly(ctx context.Context, sql string, args ...any) (out []map[string]any, cols []string, truncated bool, err error)

QueryDynamicReadOnly runs arbitrary read-only SQL for the request tenant and returns dynamic rows plus their column order. It runs inside a READ ONLY, tenant-stamped transaction, so Postgres itself rejects any write/DDL and RLS contains the reads; the backstop still guards against touching a non-RLS table without a tenant_id predicate. A statement_timeout (default 15s, tunable via RawQueryTimeout) and a row cap bound cost.

func (*Adapter) Range 

func (a *Adapter) Range(ctx context.Context, q query.RangeQuery, into any) error

Range implements query.TSQuerier for raw points over [From, To). Bucketed aggregates land with the projection phase (time_bucket).

func (*Adapter) ReadRollupWatermark added in v0.0.6 

func (a *Adapter) ReadRollupWatermark(ctx context.Context, metric string) (time.Time, bool, error)

ReadRollupWatermark reads the current (tenant, metric) rollup watermark — the newest bucket_start the maintainer has fully sealed and rolled up — from fabriq_insights_rollup_state. The query filters by tid EXPLICITLY (not solely relying on RLS to scope the read to the ctx's tenant): the rollup:insights maintainer (forgeext) enumerates tenants and runs DDL/reads through the owner/pool-path connection for some operations (e.g. TenantsForInsightsEvent, EnsureRollupTable), and a deployment where Config.Postgres.DSN itself has BYPASSRLS or superuser privileges would otherwise let this query return ANY tenant's watermark row for the metric (Postgres has no ORDER/LIMIT-independent "the right one" without an explicit filter) — silently corrupting a different tenant's sealed range and permanently excluding its historical buckets from ever being rolled up. The explicit filter makes this correct under RLS OR a bypassing role. (tenant_id, metric) is the table's PK, so there is at most one row.

Returns (zero time, false, nil) when no watermark row exists yet — the metric's first maintainer pass — rather than an error: "no watermark yet" is an expected, ordinary state, not a failure.

Uses inDynamicTenantTx (a raw driver.Tx), not inTenantTx's *pgdriver.PgTx: *pgdriver.PgTx's only scalar-read path is RawQuery.Scan, which mis-dispatches a lone struct-kind destination like *time.Time into its ORM-style struct-scan branch (resolveTable "succeeds" against time.Time with zero mapped fields, then errors with a field-count mismatch) instead of the plain scalar Scan every other single-column read here needs. driver.Tx's QueryRow(...).Scan(...) goes straight to the underlying pgx row, exactly like InsightsAdapter.QueryRaw and Query already use it.

func (*Adapter) RemoveGuardedTable added in v0.0.4 

func (a *Adapter) RemoveGuardedTable(table string)

RemoveGuardedTable unregisters a dropped dynamic table from the backstop.

func (*Adapter) RenameDynamicColumn added in v0.0.4 

func (a *Adapter) RenameDynamicColumn(ctx context.Context, table, oldName, newName string) error

RenameDynamicColumn renames a domain column on a dynamic table. The source column must be non-structural and valid; the target must be a valid, non-structural identifier.

func (*Adapter) Repair 

func (a *Adapter) Repair(ctx context.Context, tenantID string, d projection.Drift) error

Repair heals one drift through the ordinary pipeline (projection.RepairFunc):

  • missing/stale: upsert the aggregate's CURRENT state as its version's event and mark it unpublished — the relay republishes, version-gated consumers converge.
  • zombie (row gone): emit a synthetic <entity>.deleted one version past what the projection holds, so the delete applies everywhere.

Reconciliation never writes a projection engine directly.

func (*Adapter) RollupRange added in v0.0.6 

func (a *Adapter) RollupRange(ctx context.Context, m *registry.MetricSpec, from, to time.Time) error

RollupRange aggregates m's sealed events in [from, to) — grouped by scope_id (NULL preserved), the metric's declared Bucket, and its Dimensions — and upserts the result into m's rollup table, overwriting any existing row for the same (tenant, scope, bucket, dims) key (ON CONFLICT DO UPDATE). Idempotent: re-running it for the same range recomputes and overwrites the same rows with the same values.

Runs inside inTenantTx: the caller's ctx supplies the tenant (and should be unscoped — see this file's header comment — so the aggregation sees every scope's events, not just one).

func (*Adapter) SharedSchema added in v0.0.6 

func (a *Adapter) SharedSchema() string

SharedSchema reports the consolidation-mode shared schema ("" when off).

func (*Adapter) Similar 

func (a *Adapter) Similar(ctx context.Context, q query.VectorQuery, into any) error

Similar implements query.VectorQuerier: cosine nearest neighbours through the HNSW index.

func (*Adapter) SnapshotEntities 

func (a *Adapter) SnapshotEntities(ctx context.Context, tenantID string, fn func(env event.Envelope) error) error

SnapshotEntities streams every aggregate row of a tenant as synthetic <entity>.updated envelopes at the row's CURRENT version and shape — the rebuild source (projections are always rebuilt from Postgres, never from another projection). Rows are paged in id order inside stamped transactions; because appliers are pure and sinks version-gate, a row that changes mid-snapshot is healed by the live catch-up applies.

func (*Adapter) TableColumns added in v0.0.4 

func (a *Adapter) TableColumns(ctx context.Context, table string) ([]ColumnInfo, error)

TableColumns returns the physical columns of a public-schema table from information_schema, in ordinal order. A non-existent table yields an empty slice (no error). Read-only; information_schema is not tenant-scoped, so this runs on the pool without a tenant transaction.

func (*Adapter) TenantTxRaw 

func (a *Adapter) TenantTxRaw(ctx context.Context, fn func(tx *pgdriver.PgTx) error) error

TenantTxRaw opens a tenant-stamped transaction for NON-command components (e.g. the CAS index) that need raw SQL under app.tenant_id + app.scope_id. It is the only sanctioned way to run fabriq_blob_cas SQL outside the command plane; FORCE RLS isolates the work to the context tenant.

func (*Adapter) TenantsForInsightsEvent added in v0.0.6 

func (a *Adapter) TenantsForInsightsEvent(ctx context.Context, eventName string) ([]string, error)

TenantsForInsightsEvent returns the DISTINCT tenant_ids that have at least one fabriq_insights_events row named eventName (a materialized metric's Source) — the rollup:insights maintainer's per-shard tenant-enumeration seam (forgeext/rollup.go). Runs on the pool as OWNER (bypassing RLS, like TableColumns/introspect.go), not inside a tenant transaction: a maintainer pass needs to discover EVERY tenant with data for the metric on this shard/database, not one tenant's rows. This is deliberately NOT Stores.AllTenants — that seam is derived from fabriq_outbox bookkeeping (populated by the event/projection plane) and would miss any tenant that only ever called Track (Insights events bypass the outbox entirely).

func (*Adapter) Track added in v0.0.6 

func (a *Adapter) Track(ctx context.Context, events []query.AnalyticsEvent) error

Track implements query.AnalyticsQuerier — the outbox-bypass customer-event ingest. One multi-row INSERT per call; dedup_key collisions are ignored (the unique partial index on (tenant_id, dedup_key) WHERE dedup_key IS NOT NULL enforces idempotency — NULL dedup keys never conflict).

func (*Adapter) Upsert 

func (a *Adapter) Upsert(ctx context.Context, entity, id string, embedding []float32, meta map[string]any) error

Upsert implements query.VectorQuerier (kept on *Adapter for backward compat).

type CatalogStore added in v0.0.6 

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

CatalogStore is the Postgres tenant catalog (catalog.Catalog): the db-per-tenant control plane, living in a dedicated CONTROL database — never in a tenant database. It has no RLS and no tenant context; access to the control DSN is the trust boundary.

The control schema is a single table, ensured idempotently at open. (Deviation from the spec's migrations/control chain, recorded there: a one-table schema does not warrant a versioned chain yet; EnsureSchema becomes chain-managed the day it grows.)

func OpenCatalog added in v0.0.6 

func OpenCatalog(ctx context.Context, dsn string) (*CatalogStore, error)

OpenCatalog dials the control database (the writable PRIMARY) and ensures the catalog schema.

func OpenCatalogReplica added in v0.0.6 

func OpenCatalogReplica(ctx context.Context, dsn string) (*CatalogStore, error)

OpenCatalogReplica dials a read-only catalog REPLICA (a hot standby): it does NOT run schema DDL (a standby is read-only) and refuses writes. Used only as a routing-read fallback behind catalog.Failover.

func (*CatalogStore) Close added in v0.0.6 

func (s *CatalogStore) Close() error

Close releases the control connection pool.

func (*CatalogStore) Elector added in v0.0.6 

func (s *CatalogStore) Elector(key int64, opts ...ElectorOption) *Elector

Elector builds a leader elector on the CATALOG control database — the coordination point for catalog-mode singletons (the drift reconciler), which have no primary shard to elect on. Pick one key per role and keep it stable across versions.

func (*CatalogStore) Get added in v0.0.6 

func (s *CatalogStore) Get(ctx context.Context, tenantID string) (catalog.Entry, error)

Get implements catalog.Catalog.

func (*CatalogStore) List added in v0.0.6 

func (s *CatalogStore) List(ctx context.Context, cursor catalog.Cursor, limit int) ([]catalog.Entry, catalog.Cursor, error)

List implements catalog.Catalog: stable tenant-id order, keyset cursor.

func (*CatalogStore) Put added in v0.0.6 

func (s *CatalogStore) Put(ctx context.Context, e catalog.Entry) (catalog.Entry, error)

Put implements catalog.Catalog with optimistic concurrency on updated_at (zero = create).

type ClusterOps added in v0.0.6 

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

ClusterOps implements provision.ClusterOps against real Postgres clusters: CREATE DATABASE over a maintenance connection, and fabriq's migration chain over a short-lived connection to the tenant database. Both operations are idempotent — the provisioning state machine's resumability rests on it.

func NewClusterOps added in v0.0.6 

func NewClusterOps(clusterDSNs map[string]string) *ClusterOps

NewClusterOps builds ClusterOps over the configured clusters.

func (*ClusterOps) AssertBoot added in v0.0.6 

func (c *ClusterOps) AssertBoot(ctx context.Context, allowSuperuser bool) error

AssertBoot fails fast on cluster misconfiguration at Open time instead of per request (spec P6): every configured cluster must dial, and the serving credentials must not be superuser (RLS inside a tenant database does not bind superusers) unless explicitly allowed for dev/test. Clusters are checked in id order so the first error is deterministic.

func (*ClusterOps) CreateDatabase added in v0.0.6 

func (c *ClusterOps) CreateDatabase(ctx context.Context, clusterID, database string) error

CreateDatabase implements provision.ClusterOps (idempotent: an existing database is success).

func (*ClusterOps) CreateSchema added in v0.0.6 

func (c *ClusterOps) CreateSchema(ctx context.Context, clusterID, database, schema string) error

CreateSchema implements provision.SchemaClusterOps (idempotent).

func (*ClusterOps) EnsureBootstrap added in v0.0.6 

func (c *ClusterOps) EnsureBootstrap(ctx context.Context, clusterID, database, sharedSchema string) error

EnsureBootstrap implements provision.SchemaClusterOps: prepare a consolidation database ONCE with the shared schema and its extensions, so every tenant schema resolves pgvector/postgis types via search_path. Idempotent — a re-run is a cheap marker check plus IF NOT EXISTS DDL.

func (*ClusterOps) Migrate added in v0.0.6 

func (c *ClusterOps) Migrate(ctx context.Context, clusterID, database string) (string, error)

Migrate implements provision.ClusterOps: run fabriq's chain against the tenant database and report the head version.

func (*ClusterOps) MigrateSchema added in v0.0.6 

func (c *ClusterOps) MigrateSchema(ctx context.Context, clusterID, database, schema, sharedSchema string) (string, error)

MigrateSchema implements provision.SchemaClusterOps: run fabriq's chain inside a tenant schema. search_path is baked into the connection string so EVERY pooled connection resolves bare names (and grove_migrations) to the tenant schema, with the shared schema appended so the chain's CREATE EXTENSION IF NOT EXISTS steps no-op against the already-installed extensions.

func (*ClusterOps) TenantDSN added in v0.0.6 

func (c *ClusterOps) TenantDSN(clusterID, database string) (string, error)

TenantDSN derives the DSN for one tenant database on a cluster — the same derivation the catalog-mode dialer uses, exported so routing and provisioning can never disagree.

type ColumnInfo added in v0.0.4 

type ColumnInfo struct {
	Name     string `json:"name"`
	DataType string `json:"dataType"`
	Nullable bool   `json:"nullable"`
}

ColumnInfo describes one physical column of a table, read from information_schema. Used by the admin schema-drift diagnostics.

type DocStore 

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

DocStore is the Postgres document plane: an append-only update log (fabriq_crdt_updates) folded through grove's CRDT merge engine, with compacted snapshots and quiet-window materialization into ordinary entity rows + outbox events.

Update blobs are JSON-encoded []crdt.ChangeRecord (the "grove-crdt" engine named by CRDTSpec). Document ids carry their entity: "<entity>/<ulid>" — the registry's KindDocument entry binds the relational shape materialization writes.

func (*DocStore) ApplyUpdate 

func (d *DocStore) ApplyUpdate(ctx context.Context, docID string, update []byte) error

ApplyUpdate implements document.Store: append one update to the log and touch the doc's activity timestamp (the quiet-window clock).

func (*DocStore) ApplyUpdateWithSeq 

func (d *DocStore) ApplyUpdateWithSeq(ctx context.Context, docID string, update []byte) (int64, error)

ApplyUpdateWithSeq is ApplyUpdate returning the assigned log seq — the live fan-out decorator stamps it on the published sync frame so clients can detect gaps and fall back to Sync.

func (*DocStore) Compact 

func (d *DocStore) Compact(ctx context.Context, docID string) error

Compact implements document.Store: fold the log into the snapshot row and trim it, one transaction. When archive is enabled for the entity, the trimmed updates are first sealed into an immutable blob segment + index row (blob Put before the DB commit) so history survives outside the DB. Merge results never change — only their storage shape.

func (*DocStore) CompactDue added in v0.0.5 

func (d *DocStore) CompactDue(ctx context.Context) (int, error)

CompactDue compacts every unflagged document whose un-compacted update log has reached its entity's CRDTSpec.SnapshotEvery budget (<= 0 disables scheduled compaction for the entity). The budget is measured as physical per-doc log rows — seq is a table-wide identity, so watermark arithmetic cannot stand in for a count. Like MaterializeQuiet this runs from a tenant-less worker context: the bookkeeping table is scanned across tenants and the RLS'd log is counted through per-tenant transactions. Flagged docs are skipped — their raw update log is the evidence an operator resolves them with. Returns the number of documents compacted.

func (*DocStore) DeleteHistory added in v0.0.4 

func (d *DocStore) DeleteHistory(ctx context.Context, docID string) error

DeleteHistory purges a document's offloaded history: every segment blob and its index row. The GC seam for document deletion. Blob deletes run first (best-effort per key); the index rows are dropped last so a partial failure leaves recoverable pointers rather than orphaned rows.

func (*DocStore) EnableArchive added in v0.0.4 

func (d *DocStore) EnableArchive(b blob.Store, def bool)

EnableArchive wires the blob handle and the global archive default. Called by Open once the storage adapter exists; safe to call with def=false to set only the default. A nil b leaves history in Postgres.

func (*DocStore) ListSegments added in v0.0.4 

func (d *DocStore) ListSegments(ctx context.Context, docID string) ([]document.SegmentInfo, error)

ListSegments returns the sealed history segments for a document, ordered by their seq range. Blob keys are intentionally omitted (internal detail).

func (*DocStore) MaterializeQuiet 

func (d *DocStore) MaterializeQuiet(ctx context.Context, validate ValidateFunc) (int, error)

MaterializeQuiet materializes every unflagged document whose last activity is older than its entity's QuietWindow and which has updates beyond the last materialization: merged state -> validation -> entity row write + ONE <entity>.updated event (version++) through the outbox. Returns the number of documents materialized.

func (*DocStore) ReadHistory added in v0.0.4 

func (d *DocStore) ReadHistory(ctx context.Context, docID string, seqLo, seqHi int64) ([]document.HistoryUpdate, error)

ReadHistory returns every update with seqLo <= seq <= seqHi in seq order, drawn from sealed blob segments (cached after first fetch) and the live update log. A referenced segment blob that is missing is a hard error — never silently skipped (that would be undetected data loss).

func (*DocStore) Snapshot 

func (d *DocStore) Snapshot(ctx context.Context, docID string) (document.Materialized, error)

Snapshot implements document.Store: merged current state (compacted snapshot + log tail) and the materialized aggregate version.

func (*DocStore) Sync 

func (d *DocStore) Sync(ctx context.Context, docID string, stateVector []byte) ([]byte, error)

Sync implements document.Store: the state vector is an 8-byte big-endian last-seen seq (empty = from the beginning); the reply holds the compacted snapshot (when the client is behind it) plus every later update, and the new vector seq.

type Elector 

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

Elector provides advisory-lock leadership for singleton runners (the outbox relay, the reconciler). It holds pg_try_advisory_lock on a DEDICATED pooled connection (grove's ConnAcquirer), so the session-level lock cannot leak to other pool users; a connection-liveness watchdog abdicates if the session dies (the lock died with it).

fabriq-worker can therefore run any number of replicas: exactly one holds each role.

func NewElector 

func NewElector(a *Adapter, key int64, opts ...ElectorOption) *Elector

NewElector builds an elector for an advisory lock key. Pick one key per role (e.g. relay, reconciler) and keep them stable across versions.

func (*Elector) Run 

func (e *Elector) Run(ctx context.Context, lead func(ctx context.Context) error) error

Run keeps trying to lead until ctx ends. While leading it runs lead with a context that is cancelled on abdication (session loss or ctx end). lead returning (even nil) abdicates and re-campaigns.

func (*Elector) TryLead added in v0.0.6 

func (e *Elector) TryLead(ctx context.Context, lead func(ctx context.Context) error) (bool, error)

TryLead makes one non-blocking claim: if the advisory lock is free it runs lead while holding it and reports true; if another session holds the lock it reports false immediately without running lead. This is the catalog-mode sweeper's per-tenant-database work claim — many sweeper replicas race, exactly one does each tenant's pass.

type ElectorOption 

type ElectorOption func(*Elector)

ElectorOption tunes the elector.

func WithElectorHeartbeat 

func WithElectorHeartbeat(d time.Duration) ElectorOption

WithElectorHeartbeat sets the leader's session-liveness check cadence (default 5s).

func WithElectorRetry 

func WithElectorRetry(d time.Duration) ElectorOption

WithElectorRetry sets how often a non-leader retries acquisition (default 5s).

type InsightsAdapter added in v0.0.6 

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

InsightsAdapter wraps Adapter to implement query.AnalyticsQuerier.

A separate type is required because *Adapter already carries Query for query.RelationalQuerier (ctx, into any, sql string, args ...any) — the raw-SQL escape hatch behind f.Relational().Query(...) (adapter.go:649). query.AnalyticsQuerier's Query has a different shape (ctx, AnalyticsQuery, into any); Go does not allow two methods named Query with different signatures on one type, so the cube-query variant lives here. This is the same collision VectorAdapter/SpatialAdapter (vector.go, spatial.go) were already introduced to resolve for Get/Upsert — InsightsAdapter follows their exact pattern: Track has no name collision and stays delegated to the existing *Adapter method for backward compat; Query, which collides, is implemented directly here instead of on *Adapter.

QueryRaw (the read-only SQL escape hatch) rounds out the interface below.

func NewInsightsAdapter added in v0.0.6 

func NewInsightsAdapter(a *Adapter) *InsightsAdapter

NewInsightsAdapter wraps an existing Postgres adapter for the customer-facing analytics port.

func (*InsightsAdapter) EnsureRollupTable added in v0.0.6 

func (i *InsightsAdapter) EnsureRollupTable(ctx context.Context, m *registry.MetricSpec) error

EnsureRollupTable pass-through to the underlying *Adapter's owner/DDL rollup-table creation (idempotent) — see Adapter.EnsureRollupTable.

func (*InsightsAdapter) MaintainRollup added in v0.0.6 

func (i *InsightsAdapter) MaintainRollup(ctx context.Context, m *registry.MetricSpec, now time.Time) error

MaintainRollup pass-through to the underlying *Adapter's rollup maintainer pass — see Adapter.MaintainRollup. The caller supplies the tenant ctx (unscoped — MaintainRollup itself guards against a scoped ctx).

func (*InsightsAdapter) Query added in v0.0.6 

func (i *InsightsAdapter) Query(ctx context.Context, q query.AnalyticsQuery, into any) error

Query implements query.AnalyticsQuerier — on-demand cube aggregation over the tenant's fabriq_insights_events table.

This composes the pure buildInsightsSQL builder with the SAME dynamic-tx + scanMaps/assignMapsDest path List uses for map-native reads (adapter.go), rather than inTenantTx (used by Track above): *pgdriver.PgTx only exposes grove's query builder (NewSelect/NewInsert/NewUpdate/NewDelete/NewRaw), whose Scan targets a fixed struct/model type — there is no method that hands back a driver.Rows cursor for scanning into an arbitrary into. inDynamicTenantTx hands the closure a raw driver.Tx, whose Query() returns driver.Rows; scanMaps turns that into []map[string]any and assignMapsDest projects it into into, exactly as List's dynamic-entity path does.

func (*InsightsAdapter) QueryRaw added in v0.0.6 

func (i *InsightsAdapter) QueryRaw(ctx context.Context, into any, sql string, args ...any) error

QueryRaw implements query.AnalyticsQuerier — the read-only SQL escape hatch for aggregations the cube (Query, above) can't express. Modeled on (*Adapter).QueryDynamicReadOnly (adapter.go:582-617), the exact same tenant-stamped READ ONLY transaction pattern: Postgres itself rejects any write/DDL in a read-only tx, and RLS contains the reads to the caller's tenant. precheckInsightsReadOnly runs first as a cheap, fail-fast guard — defense-in-depth on top of the transaction, not a substitute for it.

This lives on *InsightsAdapter, not *Adapter, for the same reason Query does: *Adapter already has a Query method for query.RelationalQuerier with a different signature. QueryRaw has no such collision (RelationalQuerier has no QueryRaw), but it is kept here so all three AnalyticsQuerier methods live together on one receiver type.

func (*InsightsAdapter) Track added in v0.0.6 

func (i *InsightsAdapter) Track(ctx context.Context, events []query.AnalyticsEvent) error

Track implements query.AnalyticsQuerier by delegating to *Adapter.Track (no name collision there).

func (*InsightsAdapter) UpsertInsightFacts added in v0.0.6 

func (i *InsightsAdapter) UpsertInsightFacts(ctx context.Context, facts []insights.Fact) error

UpsertInsightFacts implements insights.FactSink — version-gated upsert of projected domain facts into the tenant's own fabriq_insights_facts table. Goes through i.a.inTenantTx (not the dynamic-tx path Query/QueryRaw use) so RLS contains the write to the caller's tenant, mirroring the operator sink's UpsertFacts (adapters/pganalytics/sink.go) but tenant-scoped: the tenant is derived from ctx (via inTenantTx), not carried per-fact.

type LiveStore 

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

LiveStore implements livequery.Snapshotter and livequery.Refiller over Postgres via the existing tenant-stamped read path. Ordering and the keyset boundary are Postgres-authoritative — this is the exact-top-N oracle. Reads run inside an RLS-scoped transaction (set_config app.tenant_id), so the app role only ever sees its own tenant's rows.

func (*LiveStore) After 

func (s *LiveStore) After(ctx context.Context, q livequery.LiveQuery, after livequery.Cursor, limit int) ([]livequery.Row, error)

After returns up to `limit` rows strictly after `after` in total order — the bounded keyset boundary refill.

func (*LiveStore) Members 

func (s *LiveStore) Members(ctx context.Context, q livequery.LiveQuery) ([]string, error)

Members returns every aggregate id currently matching the query's filter (tenant-scoped, RLS-enforced) — the membership seed for a Streamed subscription. No ordering or payloads; just ids.

func (*LiveStore) Snapshot 

func (s *LiveStore) Snapshot(ctx context.Context, q livequery.LiveQuery, limit int) ([]livequery.Row, error)

Snapshot returns the first `limit` rows from the anchor in total order.

type LiveSubscriptionRegistry 

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

LiveSubscriptionRegistry is the Postgres-backed durable live subscription registry (fabriq_live_subscriptions) — the backbone the sharded matcher tier uses to recover subscriptions after failover. It is worker-plane (the table has no RLS); construct it on the worker/owner connection.

func NewLiveSubscriptionRegistry 

func NewLiveSubscriptionRegistry(db *pgdriver.PgDB) *LiveSubscriptionRegistry

NewLiveSubscriptionRegistry returns the durable registry over a Postgres handle (e.g. adapter.Driver()).

func (*LiveSubscriptionRegistry) ByGateway 

func (r *LiveSubscriptionRegistry) ByGateway(ctx context.Context, gatewayID string) ([]livequery.Registration, error)

func (*LiveSubscriptionRegistry) ByPartition 

func (r *LiveSubscriptionRegistry) ByPartition(ctx context.Context, tenantID, entity string) ([]livequery.Registration, error)

func (*LiveSubscriptionRegistry) ByPartitionNum 

func (r *LiveSubscriptionRegistry) ByPartitionNum(ctx context.Context, p int) ([]livequery.Registration, error)

func (*LiveSubscriptionRegistry) Delete 

func (r *LiveSubscriptionRegistry) Delete(ctx context.Context, subID string) error

func (*LiveSubscriptionRegistry) Put 

func (r *LiveSubscriptionRegistry) Put(ctx context.Context, reg livequery.Registration) error

type Maintenance added in v0.0.6 

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

Maintenance is one tenant database's single-pass worker surface (spec 2026-07-03 D5): the claim-guarded form of the loops the static worker runs boot-time. Each Sweep try-claims the database's own advisory locks, so any number of sweeper replicas cooperate — losers skip cleanly.

func NewMaintenance added in v0.0.6 

func NewMaintenance(a *Adapter, reg *registry.Registry, pub event.Publisher, docs *DocStore) *Maintenance

NewMaintenance builds the maintenance surface for one tenant database. docs MUST be the SAME DocStore the serving plane uses (archive-enabled when history offload is configured) — minting a fresh a.Documents() here would silently skip history sealing on compaction. pub may be nil (no event transport): the relay phase is skipped and outbox rows accumulate.

func (*Maintenance) Sweep added in v0.0.6 

func (m *Maintenance) Sweep(ctx context.Context, compact bool) (sweep.Result, error)

Sweep implements sweep.Maintainer: one maintenance pass over this database — relay the outbox, materialize quiet documents, and (when compact is set) compact due document logs. Lock losers report Claimed=false with whatever the won phases did. Electors are minted per pass so their advisory-lock keys reflect the ctx schema (schema mode).

type Option 

type Option func(*openConfig)

Option configures Open.

func WithGuardedTables 

func WithGuardedTables(tables ...string) Option

WithGuardedTables adds tables to the tenant guard beyond the registry's (e.g. the telemetry hypertable, which has no RLS because Timescale columnstore forbids it).

func WithPoolSize 

func WithPoolSize(n int) Option

WithPoolSize sets the pgx pool size.

func WithSharedSchema added in v0.0.6 

func WithSharedSchema(name string) Option

WithSharedSchema enables schema-per-tenant stamping: when a tenant schema is present on ctx (pathctx), every stamped transaction sets search_path = "<tenant_schema>, <shared>" so bare table names resolve to the tenant's schema and extension types resolve from the shared schema. Empty (the default) means no search_path is ever stamped — legacy behavior.

type Relay 

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

Relay is the outbox relay: it drains unpublished outbox rows (FOR UPDATE SKIP LOCKED, ULID order) and publishes them through an event.Publisher, woken by LISTEN/NOTIFY with interval polling as the safety net.

Delivery is at-least-once: rows are published before being marked, so a crash between the two replays the event; consumers are version-gated idempotent by contract. Run exactly one active relay (wrap Run in an Elector) — multiple relays are safe (SKIP LOCKED) but waste publishes.

func NewRelay 

func NewRelay(a *Adapter, reg *registry.Registry, pub event.Publisher, opts ...RelayOption) *Relay

NewRelay builds a relay on the adapter's pool.

func (*Relay) Backlog 

func (r *Relay) Backlog(ctx context.Context) (int64, error)

Backlog reports the unpublished outbox depth (metrics).

func (*Relay) DrainAll added in v0.0.6 

func (r *Relay) DrainAll(ctx context.Context) (int, error)

DrainAll drains the outbox until empty (batches of the configured size) and returns the number of envelopes published — the single-pass form the catalog-mode sweeper runs under its per-tenant-database claim, where Run's LISTEN/NOTIFY loop would hold a connection per tenant forever.

func (*Relay) Run 

func (r *Relay) Run(ctx context.Context) error

Run drains until ctx ends.

type RelayOption 

type RelayOption func(*Relay)

RelayOption tunes the relay.

func WithRelayBatch 

func WithRelayBatch(n int) RelayOption

WithRelayBatch sets the per-transaction drain batch (default 128).

func WithRelayOnPublish 

func WithRelayOnPublish(fn func(n int)) RelayOption

WithRelayOnPublish installs a per-batch callback (metrics).

func WithRelayPollInterval 

func WithRelayPollInterval(d time.Duration) RelayOption

WithRelayPollInterval sets the fallback poll cadence (default 1s; NOTIFY normally wakes the relay first).

type SpatialAdapter 

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

SpatialAdapter wraps Adapter to implement query.SpatialQuerier. A separate type is required because *Adapter already carries Upsert for query.VectorQuerier ([]float32 embedding) — Go does not allow two methods with the same name on one type, so the spatial variant lives here.

func NewSpatialAdapter 

func NewSpatialAdapter(a *Adapter) *SpatialAdapter

NewSpatialAdapter wraps an existing Postgres adapter for geometry operations.

func (*SpatialAdapter) Covering added in v0.0.6 

func (s *SpatialAdapter) Covering(ctx context.Context, q query.CoverQuery, into any) error

Covering implements query.SpatialCoverer: a GiST-accelerated topological containment search (no radius). Containment is boolean, so DistanceM is 0 and results are unordered beyond the K cap. SRID mixing is the caller's responsibility — the probe and stored geometry must share an SRID for PostGIS to compare them (unlike Within, no geography cast is applied).

func (*SpatialAdapter) Delete 

func (s *SpatialAdapter) Delete(ctx context.Context, entity, id string) error

Delete implements query.SpatialQuerier.

func (*SpatialAdapter) Get added in v0.0.5 

func (s *SpatialAdapter) Get(ctx context.Context, entity, id string) (geom query.Geometry, meta map[string]any, ok bool, err error)

Get implements query.SpatialQuerier: fetch the stored geometry (as WKT) and meta for (tenant, entity, id). ok=false (nil error) when the id has no row.

func (*SpatialAdapter) Upsert 

func (s *SpatialAdapter) Upsert(ctx context.Context, entity, id string, geom query.Geometry, meta map[string]any) error

Upsert implements query.SpatialQuerier: store/replace a geometry from WKT+SRID.

func (*SpatialAdapter) Within 

func (s *SpatialAdapter) Within(ctx context.Context, q query.SpatialQuery, into any) error

Within implements query.SpatialQuerier: GiST-accelerated radius search, nearest-first. For SRID 4326 distance/predicate use the geography cast (true metres); otherwise planar metres in the geometry's own units.

type StateRepo 

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

StateRepo is the Postgres-backed projection bookkeeping (worker-plane tables, no RLS — consumers and the reconciler are cross-tenant by design). It implements projection.StateRepo.

func (*StateRepo) AppliedVersion 

func (r *StateRepo) AppliedVersion(ctx context.Context, tenantID, proj, aggregate, aggID string) (int64, error)

AppliedVersion implements projection.StateReader.

func (*StateRepo) Get 

func (r *StateRepo) Get(ctx context.Context, tenantID, proj string) (projection.State, error)

Get implements projection.StateRepo.

func (*StateRepo) SetApplied 

func (r *StateRepo) SetApplied(ctx context.Context, tenantID, proj, aggregate, aggID string, version int64) error

SetApplied records a projection apply; the watermark never regresses.

func (*StateRepo) Tenants 

func (r *StateRepo) Tenants(ctx context.Context) ([]string, error)

Tenants lists every tenant that has ever emitted an event (worker-plane discovery for rebuild --all-tenants and the reconciler; the outbox has no RLS, so this sees across tenants by design).

func (*StateRepo) Upsert 

func (r *StateRepo) Upsert(ctx context.Context, s projection.State) error

Upsert implements projection.StateRepo.

type ValidateFunc 

type ValidateFunc func(entity string, vals map[string]any) error

ValidateFunc is the post-merge validation hook: CRDTs converge but do not guarantee business validity. A non-nil error flags the document for resolution instead of materializing.

type VectorAdapter added in v0.0.3 

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

VectorAdapter wraps Adapter to implement query.VectorQuerier. A separate type is required because *Adapter already carries Get for query.RelationalQuerier (entity, id string, into any) — Go does not allow two methods with the same name on one type, so the vector variant lives here. The existing Upsert/Similar/Delete methods remain on *Adapter for backwards compat; VectorAdapter delegates to them.

func NewVectorAdapter added in v0.0.3 

func NewVectorAdapter(a *Adapter) *VectorAdapter

NewVectorAdapter wraps an existing Postgres adapter for vector operations.

func (*VectorAdapter) Delete added in v0.0.3 

func (v *VectorAdapter) Delete(ctx context.Context, entity, id string) error

Delete implements query.VectorQuerier.

func (*VectorAdapter) DeleteByMeta added in v0.0.3 

func (v *VectorAdapter) DeleteByMeta(ctx context.Context, entity string, filter map[string]string) error

DeleteByMeta implements query.VectorQuerier.

func (*VectorAdapter) Get added in v0.0.3 

func (v *VectorAdapter) Get(ctx context.Context, entity, id string) ([]float32, error)

Get implements query.VectorQuerier. Returns the stored embedding for (entity, id) as []float32, or *fabriqerr.NotFoundError on miss.

func (*VectorAdapter) Similar added in v0.0.3 

func (v *VectorAdapter) Similar(ctx context.Context, q query.VectorQuery, into any) error

Similar implements query.VectorQuerier.

func (*VectorAdapter) Upsert added in v0.0.3 

func (v *VectorAdapter) Upsert(ctx context.Context, entity, id string, embedding []float32, meta map[string]any) error

Upsert implements query.VectorQuerier.

Source Files

View all Source files

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.