forma

package module
v0.1.0 Latest Latest
Warning

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

 Go to latest Published: Jun 20, 2026 License: MIT Imports: 7 Imported by: 0

README

Forma

Forma is a general-purpose data management system built on PostgreSQL. It uses JSON Schema for data definition and a dual storage model (Hot Fields Table + EAV Table) to handle highly dynamic data structures without schema migrations.

Prerequisites

  • Go 1.26+
  • Docker & Docker Compose (for local PostgreSQL and S3-compatible storage)
  • Bun (for E2E test scripts)
  • k6 (for load testing; Docker fallback available)

Quick Start

# Clone and enter the project
git clone https://github.com/lychee-technology/forma.git
cd forma

# Start PostgreSQL via Docker Compose
docker compose -f deploy/docker-compose.yml up -d

# Build all binaries
make build-all

# Initialize database tables (required once)
./build/tools init-db \
  --db-host localhost \
  --db-port 5432 \
  --db-name forma \
  --db-user postgres \
  --db-password postgres \
  --db-ssl-mode disable \
  --schema-dir cmd/server/schemas

# Start the server
SCHEMA_DIR=cmd/server/schemas ./build/server

Or use the convenience script that does all of the above:

./scripts/local_server.sh

The server listens on port 8080 by default. Configure via environment variables:

Variable Default Description
DB_HOST localhost PostgreSQL host
DB_PORT 5432 PostgreSQL port
DB_NAME forma Database name
DB_USER postgres Database user
DB_PASSWORD `` Database password
DB_SSL_MODE disable SSL mode
SCHEMA_DIR `` Directory containing schema JSON files
PORT 8080 HTTP listen port

API Reference

Method Path Description
POST /api/v1/{schema} Create records (single object or array)
GET /api/v1/{schema}/{row_id} Get a single record
GET /api/v1/{schema} Query records with pagination (?page=&items_per_page=&sort_by=&sort_order=&attrs=)
PUT /api/v1/{schema}/{row_id} Update a record
DELETE /api/v1/{schema} Batch delete (JSON body: array of row_id strings)
GET /api/v1/search Cross-schema search (?schemas=&q=&page=&items_per_page=)
POST /api/v1/advanced_query Advanced query with condition DSL (JSON body)

Testing

Unit & Integration Tests
# Run all unit and integration tests
make test

# Run with coverage report
make coverage

# Run linter
make lint
Go E2E Harness (container-based)

Requires Docker. Validates the three-tier federated query architecture (Postgres Hot + S3 Delta/Base → DuckDB merge-on-read).

# Smoke test: verify infrastructure starts
go test -v ./internal/e2e_harness/... -timeout=5m

# Full federated suite (functional + consistency + failure modes)
go test -v ./internal/e2e_harness/federated/... -tags=e2e -timeout=30m

# Performance tests only (longer timeout)
go test -v ./internal/e2e_harness/federated/... -run TestPerformance -tags=e2e -timeout=60m
Bun E2E (black-box API validation)

Requires a running Forma server and PostgreSQL.

cd tests/e2e
cp .env.example .env
bun install

# Default pipeline: register schemas → generate data → CDC flush → federated check
bun run test

# Individual steps
bun run register-schemas
bun run gen-data -- --schema all --count 10000
bun run cdc-flush
bun run federated-check

# Extended steps
bun run cdc-init          # Backfill base parquet
bun run compactor -- --all # Merge delta into base
k6 Load Testing
cd tests/e2e
bun run build-k6

bun run k6-smoke   # 5 VUs, 30s
bun run k6-full    # 30 VUs, 2m
bun run k6-perf    # 100 VUs, 5m
Benchmarks
make benchmark-smoke       # CI smoke validation
make benchmark-regression  # Small live subset
make benchmark-heavy       # Heavy planning set

Documentation

Why Forma?

Forma targets the gap between rigid RDBMS schemas and schema-less NoSQL stores:

  • Zero-Downtime Schema Evolution — Add or modify fields by updating JSON Schema metadata; no ALTER TABLE required.
  • ACID on PostgreSQL — Inherits full transactional guarantees from Postgres.
  • Smart SQL Generation — CTE + JSON_AGG eliminates N+1 queries in EAV models.
  • Federated Query (Lakehouse) — PostgreSQL for OLTP, DuckDB + Parquet on S3 for OLAP. Anti-Join + Dirty Set ensures consistency across tiers.

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var ErrConflict = errors.New("conflict")

ErrConflict is returned when an operation would violate a uniqueness constraint or produce a duplicate record.

View Source 
var ErrInvalidInput = errors.New("invalid input")

ErrInvalidInput is returned when caller-supplied input fails validation (missing required fields, unsupported values, etc.).

View Source 
var ErrNotFound = errors.New("not found")

ErrNotFound is returned when a requested entity does not exist.

Functions

This section is empty.

Types

type AttributeMetadata added in v0.0.3 

type AttributeMetadata struct {
	AttributeName  string         `json:"attr_name"`  // attr_name, JSON Path
	AttributeID    int16          `json:"attr_id"`    // attr_id
	ValueType      ValueType      `json:"value_type"` // 'text', 'numeric', 'date', 'bool'
	RequiredPolicy RequiredPolicy `json:"required_policy,omitempty"`
	// Required is kept for backward compatibility with older metadata readers.
	// New code should use RequiredPolicy.
	Required      bool               `json:"required,omitempty"`
	ColumnBinding *MainColumnBinding `json:"column_binding,omitempty"`
}

AttributeMetadata stores cached metadata from the attributes table.

func (AttributeMetadata) EffectiveRequiredPolicy added in v0.0.27 

func (m AttributeMetadata) EffectiveRequiredPolicy() RequiredPolicy

EffectiveRequiredPolicy returns the policy used at runtime. If RequiredPolicy is unset, it falls back to legacy Required semantics.

func (AttributeMetadata) IsInsideArray added in v0.0.3 

func (m AttributeMetadata) IsInsideArray() bool

IsInsideArray infers if the attribute is inside an array based on its name.

func (AttributeMetadata) Location added in v0.0.3 

func (m AttributeMetadata) Location() AttributeStorageLocation

Location returns where the attribute is stored. If ColumnBinding is nil, the attribute is in EAV; otherwise in main table.

type AttributeStorageLocation added in v0.0.3 

type AttributeStorageLocation string

AttributeStorageLocation enumerates where the attribute physically resides. 

const (
	AttributeStorageLocationUnknown AttributeStorageLocation = ""
	AttributeStorageLocationMain    AttributeStorageLocation = "main"
	AttributeStorageLocationEAV     AttributeStorageLocation = "eav"
)

type BatchConfig added in v0.0.2 

type BatchConfig struct {
	EnableDynamicSizing      bool `json:"enableDynamicSizing"`
	EnableParallelProcessing bool `json:"enableParallelProcessing"`
	EnableBatchStreaming     bool `json:"enableBatchStreaming"`
	ParallelThreshold        int  `json:"parallelThreshold"`
	StreamingThreshold       int  `json:"streamingThreshold"`
	MaxParallelWorkers       int  `json:"maxParallelWorkers"`
	StreamingChunkSize       int  `json:"streamingChunkSize"`
	StreamingDelay           int  `json:"streamingDelay"` // milliseconds
	MaxComplexityPerBatch    int  `json:"maxComplexityPerBatch"`
	AttributeComplexityScore int  `json:"attributeComplexityScore"`
	OptimalChunkSize         int  `json:"optimalChunkSize"`
}

BatchConfig contains batch processing settings

type BatchOperation 

type BatchOperation struct {
	Operations []EntityOperation `json:"operations"`
	Atomic     bool              `json:"atomic"` // Request all-or-nothing execution; may be rejected when unsupported.
}

BatchOperation represents batch entity operations

type BatchResult 

type BatchResult struct {
	Successful []*DataRecord    `json:"successful"`
	Failed     []OperationError `json:"failed"`
	TotalCount int              `json:"totalCount"`
	Duration   int64            `json:"duration"` // microseconds
}

BatchResult represents results from batch operations

type CascadeAction added in v0.0.2 

type CascadeAction string

CascadeAction defines the type of cascade action 

const (
	CascadeActionDelete   CascadeAction = "delete"
	CascadeActionUpdate   CascadeAction = "update"
	CascadeActionNullify  CascadeAction = "nullify"
	CascadeActionRestrict CascadeAction = "restrict"
)

type CascadeRule added in v0.0.2 

type CascadeRule struct {
	SourceSchema string        `json:"sourceSchema"`
	TargetSchema string        `json:"targetSchema"`
	Action       CascadeAction `json:"action"`
	MaxDepth     int           `json:"maxDepth,omitempty"`
}

CascadeRule defines cascade behavior for specific schema relationships

type CompositeCondition 

type CompositeCondition struct {
	Logic      Logic       `json:"l"`
	Conditions []Condition `json:"c"`
}

--- 3. Composite Condition (Non-Leaf Node) ---

func (*CompositeCondition) IsLeaf 

func (c *CompositeCondition) IsLeaf() bool

func (*CompositeCondition) UnmarshalJSON 

func (c *CompositeCondition) UnmarshalJSON(data []byte) error

UnmarshalJSON customizes decoding so that nested conditions are turned into the appropriate concrete condition implementations.

type Condition 

type Condition interface {
	IsLeaf() bool
}

--- 2. Interface (The Core) ---

type Config added in v0.0.2 

type Config struct {
	Database       DatabaseConfig    `json:"database"`
	Query          QueryConfig       `json:"query"`
	Entity         EntityConfig      `json:"entity"`
	Transaction    TransactionConfig `json:"transaction"`
	Performance    PerformanceConfig `json:"performance"`
	Logging        LoggingConfig     `json:"logging"`
	Metrics        MetricsConfig     `json:"metrics"`
	Reference      ReferenceConfig   `json:"reference"`
	DuckDB         DuckDBConfig      `json:"duckdb"`
	SchemaRegistry SchemaRegistry    `json:"-"` // Custom schema registry implementation (optional)
}

Config consolidates settings from both modules

func DefaultConfig added in v0.0.2 

func DefaultConfig(schemaRegistry SchemaRegistry) *Config

DefaultConfig returns a default configuration

func NewConfig added in v0.0.25 

func NewConfig(opts ...Option) *Config

NewConfig creates a Config starting from the defaults produced by DefaultConfig(nil) and then applies each provided Option in order. The schema registry can be set via WithSchemaRegistry.

Example: 

cfg := forma.NewConfig(
    forma.WithDatabase(forma.DatabaseConfig{Host: "db.example.com", Port: 5432, MaxConnections: 50}),
    forma.WithDuckDB(forma.DuckDBConfig{Enabled: true, DBPath: ":memory:"}),
)

func (*Config) Validate added in v0.0.2 

func (c *Config) Validate() error

Validate validates the configuration

type ConfigError added in v0.0.2 

type ConfigError struct {
	Field   string `json:"field"`
	Message string `json:"message"`
}

ConfigError represents a configuration validation error

func (*ConfigError) Error added in v0.0.2 

func (e *ConfigError) Error() string

type CrossSchemaRequest 

type CrossSchemaRequest struct {
	SchemaNames  []string  `json:"schema_names" validate:"required"`
	SearchTerm   string    `json:"search_term" validate:"required"`
	Page         int       `json:"page" validate:"min=1"`
	ItemsPerPage int       `json:"items_per_page" validate:"min=1,max=100"`
	Condition    Condition `json:"-"`               // Custom unmarshal, can be CompositeCondition or KvCondition
	Attrs        []string  `json:"attrs,omitempty"` // Attributes to return (field projection)
}

CrossSchemaRequest represents a cross-schema search request

func (CrossSchemaRequest) MarshalJSON 

func (r CrossSchemaRequest) MarshalJSON() ([]byte, error)

MarshalJSON implements custom JSON marshaling for CrossSchemaRequest.

func (*CrossSchemaRequest) UnmarshalJSON 

func (r *CrossSchemaRequest) UnmarshalJSON(data []byte) error

UnmarshalJSON implements custom JSON unmarshaling for CrossSchemaRequest. It allows the Condition field to be either a CompositeCondition or KvCondition.

type CursorQueryResult 

type CursorQueryResult struct {
	Data          []*DataRecord `json:"data"`
	NextCursor    string        `json:"next_cursor,omitempty"`
	HasMore       bool          `json:"has_more"`
	ExecutionTime time.Duration `json:"execution_time"`
}

CursorQueryResult represents cursor-based pagination results.

type DataRecord 

type DataRecord struct {
	SchemaName string         `json:"schema_name"`
	RowID      uuid.UUID      `json:"row_id"`
	Attributes map[string]any `json:"attributes"`
}

type DatabaseConfig added in v0.0.2 

type DatabaseConfig struct {
	Host            string        `json:"host"`
	Port            int           `json:"port"`
	Database        string        `json:"database"`
	Username        string        `json:"username"`
	Password        string        `json:"password"`
	SSLMode         string        `json:"sslMode"`
	Schema          string        `json:"schema"`
	MaxConnections  int           `json:"maxConnections"`
	MaxIdleConns    int           `json:"maxIdleConns"`
	ConnMaxLifetime time.Duration `json:"connMaxLifetime"`
	ConnMaxIdleTime time.Duration `json:"connMaxIdleTime"`
	Timeout         time.Duration `json:"timeout"`
	TableNames      TableNames    `json:"tableNames"`
}

DatabaseConfig contains database connection settings

type DuckDBConfig added in v0.0.23 

type DuckDBConfig struct {
	Enabled                 bool          `json:"enabled"`
	DBPath                  string        `json:"dbPath"`        // path to local DuckDB file (or ":memory:")
	MemoryLimitMB           int           `json:"memoryLimitMB"` // memory limit for DuckDB in MB
	EnableS3                bool          `json:"enableS3"`      // enable S3/http file system
	S3Endpoint              string        `json:"s3Endpoint"`    // custom S3 endpoint (for MinIO)
	S3AccessKey             string        `json:"s3AccessKey"`
	S3SecretKey             string        `json:"s3SecretKey"`
	S3Region                string        `json:"s3Region"`
	EnableParquet           bool          `json:"enableParquet"` // enable parquet extension
	Extensions              []string      `json:"extensions"`    // additional extensions to load
	MaxConnections          int           `json:"maxConnections"`
	QueryTimeout            time.Duration `json:"queryTimeout"`            // per-query timeout for DuckDB access
	MaxParallelism          int           `json:"maxParallelism"`          // max threads/pragmas for DuckDB
	CircuitBreakerThreshold float64       `json:"circuitBreakerThreshold"` // failure rate to trip circuit breaker (0..1)
	Routing                 RoutingPolicy `json:"routing"`                 // routing policy for federated queries
}

DuckDBConfig contains DuckDB connection and S3 settings for federated queries

type EntityBatchCreator added in v0.0.24 

type EntityBatchCreator interface {
	BatchCreate(ctx context.Context, req *BatchOperation) (*BatchResult, error)
}

type EntityBatchOperator added in v0.0.24 

type EntityBatchOperator interface {
	BatchCreate(ctx context.Context, req *BatchOperation) (*BatchResult, error)
	BatchUpdate(ctx context.Context, req *BatchOperation) (*BatchResult, error)
	BatchDelete(ctx context.Context, req *BatchOperation) (*BatchResult, error)
}

type EntityConfig added in v0.0.2 

type EntityConfig struct {
	EnableReferenceValidation bool          `json:"enableReferenceValidation"`
	EnableCascadeDelete       bool          `json:"enableCascadeDelete"`
	BatchSize                 int           `json:"batchSize"`
	CacheEnabled              bool          `json:"cacheEnabled"`
	CacheTTL                  time.Duration `json:"cacheTTL"`
	MaxEntitySize             int           `json:"maxEntitySize"`
	EnableVersioning          bool          `json:"enableVersioning"`
	SchemaDirectory           string        `json:"schemaDirectory"`
}

EntityConfig contains entity management settings

type EntityIdentifier 

type EntityIdentifier struct {
	SchemaName string    `json:"schemaName"`
	RowID      uuid.UUID `json:"rowId"`
}

EntityIdentifier identifies an entity for operations

type EntityManager 

type EntityManager interface {
	EntityWriter
	EntityReader
	EntityBatchOperator
}

EntityManager provides comprehensive entity and query operations

type EntityOperation 

type EntityOperation struct {
	EntityIdentifier
	Type    OperationType  `json:"type"`
	Data    map[string]any `json:"data,omitempty"`
	Updates map[string]any `json:"updates,omitempty"`
}

EntityOperation represents CRUD operations

type EntityReader added in v0.0.24 

type EntityReader interface {
	Get(ctx context.Context, req *QueryRequest) (*DataRecord, error)
	Query(ctx context.Context, req *QueryRequest) (*QueryResult, error)
	CrossSchemaSearch(ctx context.Context, req *CrossSchemaRequest) (*QueryResult, error)
}

type EntityUpdate 

type EntityUpdate struct {
	EntityIdentifier
	Updates any `json:"updates"`
}

EntityUpdate represents an update operation

type EntityWriter added in v0.0.24 

type EntityWriter interface {
	Create(ctx context.Context, req *EntityOperation) (*DataRecord, error)
	Update(ctx context.Context, req *EntityOperation) (*DataRecord, error)
	Delete(ctx context.Context, req *EntityOperation) error
}

type FederatedQueryRequest added in v0.1.0 

type FederatedQueryRequest struct {
	Enabled                  bool     `json:"enabled,omitempty"`
	PreferredTiers           []string `json:"preferred_tiers,omitempty"`
	PreferHot                bool     `json:"prefer_hot,omitempty"`
	UseMainAsAnchor          bool     `json:"use_main_as_anchor,omitempty"`
	S3ParquetPathTemplate    string   `json:"s3_parquet_path_template,omitempty"`
	AllowPartialDegradedMode bool     `json:"allow_partial_degraded_mode,omitempty"`
	IncludeExecutionPlan     bool     `json:"include_execution_plan,omitempty"`
	ConsistencyMode          string   `json:"consistency_mode,omitempty"`
}

FederatedQueryRequest carries optional hints for routing QueryRequest through the federated repository path while preserving the normal API surface for callers that do not need DuckDB/S3-backed reads.

type FilterField 

type FilterField string
const (
	FilterFieldAttributeName FilterField = "attr_name"
	FilterFieldValueText     FilterField = "value_text"
	FilterFieldValueNumeric  FilterField = "value_numeric"
	FilterFieldRowID         FilterField = "row_id"
	FilterFieldSchemaName    FilterField = "schema_name"
)

type FilterType 

type FilterType string

FilterType defines supported filter operations 

const (
	FilterEquals      FilterType = "equals"
	FilterNotEquals   FilterType = "not_equals"
	FilterStartsWith  FilterType = "starts_with"
	FilterContains    FilterType = "contains"
	FilterGreaterThan FilterType = "gt"
	FilterLessThan    FilterType = "lt"
	FilterGreaterEq   FilterType = "gte"
	FilterLessEq      FilterType = "lte"
	FilterIn          FilterType = "in"
	FilterNotIn       FilterType = "not_in"
)

type JSONSchema added in v0.0.15 

type JSONSchema struct {
	ID         int16                      `json:"id"`
	Name       string                     `json:"name"`
	Version    int                        `json:"version"`
	Schema     string                     `json:"schema"`
	Properties map[string]*PropertySchema `json:"properties"`
	Required   []string                   `json:"required"`
	CreatedAt  int64                      `json:"created_at"`
}

JSONSchema represents a schema definition.

type KvCondition 

type KvCondition struct {
	Attr  string `json:"a"`
	Value string `json:"v"`
}

--- 4. KvCondition (Leaf Node) ---

func (*KvCondition) IsLeaf 

func (kv *KvCondition) IsLeaf() bool

func (*KvCondition) UnmarshalJSON 

func (kv *KvCondition) UnmarshalJSON(data []byte) error

UnmarshalJSON ensures short-hand keys are present.

type LoggingConfig added in v0.0.2 

type LoggingConfig struct {
	Level                  string        `json:"level"`
	Format                 string        `json:"format"`
	EnableStructured       bool          `json:"enableStructured"`
	EnablePerformance      bool          `json:"enablePerformance"`
	EnableQueryLogging     bool          `json:"enableQueryLogging"`
	LogSlowQueries         bool          `json:"logSlowQueries"`
	SlowQueryThreshold     time.Duration `json:"slowQueryThreshold"`
	MaxLogSize             int           `json:"maxLogSize"`
	LogRotation            bool          `json:"logRotation"`
	SanitizeParameters     bool          `json:"sanitizeParameters"`
	LogQueries             bool          `json:"logQueries"`
	LogErrors              bool          `json:"logErrors"`
	LogSecurityEvents      bool          `json:"logSecurityEvents"`
	LogPerformanceWarnings bool          `json:"logPerformanceWarnings"`
	LogAllOperations       bool          `json:"logAllOperations"`
	EnableDetailedLogging  bool          `json:"enableDetailedLogging"`
}

LoggingConfig contains logging settings

type Logic 

type Logic string
const (
	LogicAnd Logic = "and"
	LogicOr  Logic = "or"
)

type MainColumn added in v0.0.3 

type MainColumn string

MainColumn represents column names in the main entity table. 

const (
	MainColumnText01     MainColumn = "text_01"
	MainColumnText02     MainColumn = "text_02"
	MainColumnText03     MainColumn = "text_03"
	MainColumnText04     MainColumn = "text_04"
	MainColumnText05     MainColumn = "text_05"
	MainColumnText06     MainColumn = "text_06"
	MainColumnText07     MainColumn = "text_07"
	MainColumnText08     MainColumn = "text_08"
	MainColumnText09     MainColumn = "text_09"
	MainColumnText10     MainColumn = "text_10"
	MainColumnSmallint01 MainColumn = "smallint_01"
	MainColumnSmallint02 MainColumn = "smallint_02"
	MainColumnInteger01  MainColumn = "integer_01"
	MainColumnInteger02  MainColumn = "integer_02"
	MainColumnInteger03  MainColumn = "integer_03"
	MainColumnBigint01   MainColumn = "bigint_01"
	MainColumnBigint02   MainColumn = "bigint_02"
	MainColumnBigint03   MainColumn = "bigint_03"
	MainColumnBigint04   MainColumn = "bigint_04"
	MainColumnBigint05   MainColumn = "bigint_05"
	MainColumnDouble01   MainColumn = "double_01"
	MainColumnDouble02   MainColumn = "double_02"
	MainColumnDouble03   MainColumn = "double_03"
	MainColumnDouble04   MainColumn = "double_04"
	MainColumnDouble05   MainColumn = "double_05"
	MainColumnUUID01     MainColumn = "uuid_01"
	MainColumnUUID02     MainColumn = "uuid_02"
	MainColumnCreatedAt  MainColumn = "ltbase_created_at"
	MainColumnUpdatedAt  MainColumn = "ltbase_updated_at"
	MainColumnDeletedAt  MainColumn = "ltbase_deleted_at"
	MainColumnCreatedBy  MainColumn = "ltbase_created_by"
	MainColumnUpdatedBy  MainColumn = "ltbase_updated_by"
	MainColumnDeletedBy  MainColumn = "ltbase_deleted_by"
	MainColumnSchemaID   MainColumn = "ltbase_schema_id"
	MainColumnRowID      MainColumn = "ltbase_row_id"
)

type MainColumnBinding added in v0.0.3 

type MainColumnBinding struct {
	ColumnName MainColumn         `json:"col_name"`
	Encoding   MainColumnEncoding `json:"encoding,omitempty"`
}

MainColumnBinding describes how a schema attribute maps into a hot attribute column.

func (*MainColumnBinding) ColumnType added in v0.0.3 

func (m *MainColumnBinding) ColumnType() MainColumnType

ColumnType derives the column type from the column name prefix.

type MainColumnEncoding added in v0.0.3 

type MainColumnEncoding string

MainColumnEncoding represents special encoding for main column values. 

const (
	MainColumnEncodingDefault  MainColumnEncoding = "default"
	MainColumnEncodingBoolText MainColumnEncoding = "bool_text" // "1"/"0" string in a text column
	MainColumnEncodingUnixMs   MainColumnEncoding = "unix_ms"
	MainColumnEncodingBoolInt  MainColumnEncoding = "bool_smallint"
	MainColumnEncodingISO8601  MainColumnEncoding = "iso8601"
)

type MainColumnType added in v0.0.3 

type MainColumnType string

MainColumnType represents the data type of a main column. 

const (
	MainColumnTypeText     MainColumnType = "text"
	MainColumnTypeSmallint MainColumnType = "smallint"
	MainColumnTypeInteger  MainColumnType = "integer"
	MainColumnTypeBigint   MainColumnType = "bigint"
	MainColumnTypeDouble   MainColumnType = "double"
	MainColumnTypeUUID     MainColumnType = "uuid"
)

type MetricsConfig added in v0.0.2 

type MetricsConfig struct {
	Enabled                  bool              `json:"enabled"`
	Provider                 string            `json:"provider"` // prometheus, statsd, etc.
	Endpoint                 string            `json:"endpoint"`
	CollectionInterval       time.Duration     `json:"collectionInterval"`
	EnableHistograms         bool              `json:"enableHistograms"`
	EnableCounters           bool              `json:"enableCounters"`
	EnableGauges             bool              `json:"enableGauges"`
	Namespace                string            `json:"namespace"`
	Labels                   map[string]string `json:"labels"`
	MaxSamples               int               `json:"maxSamples"`
	EnableOperationMetrics   bool              `json:"enableOperationMetrics"`
	EnableTransactionMetrics bool              `json:"enableTransactionMetrics"`
	EnablePatternMetrics     bool              `json:"enablePatternMetrics"`
}

MetricsConfig contains metrics collection settings

type OperationError 

type OperationError struct {
	Operation EntityOperation `json:"operation"`
	Error     string          `json:"error"`
	Code      string          `json:"code"`
	Details   map[string]any  `json:"details,omitempty"`
}

OperationError represents an error for a specific operation

type OperationType 

type OperationType string

OperationType represents CRUD operations 

const (
	OperationCreate OperationType = "create"
	OperationRead   OperationType = "read"
	OperationUpdate OperationType = "update"
	OperationDelete OperationType = "delete"
	OperationQuery  OperationType = "query"
)

type Option added in v0.0.25 

type Option func(*Config)

Option is a functional option that mutates a Config. Use the With* constructors below to build option values, then pass them to NewConfig to obtain a fully-configured Config without touching the struct fields directly.

func WithDatabase added in v0.0.25 

func WithDatabase(db DatabaseConfig) Option

WithDatabase replaces the DatabaseConfig section.

func WithDuckDB added in v0.0.25 

func WithDuckDB(d DuckDBConfig) Option

WithDuckDB replaces the DuckDBConfig section.

func WithEntity added in v0.0.25 

func WithEntity(e EntityConfig) Option

WithEntity replaces the EntityConfig section.

func WithLogging added in v0.0.25 

func WithLogging(l LoggingConfig) Option

WithLogging replaces the LoggingConfig section.

func WithMetrics added in v0.0.25 

func WithMetrics(m MetricsConfig) Option

WithMetrics replaces the MetricsConfig section.

func WithPerformance added in v0.0.25 

func WithPerformance(p PerformanceConfig) Option

WithPerformance replaces the PerformanceConfig section.

func WithQuery added in v0.0.25 

func WithQuery(q QueryConfig) Option

WithQuery replaces the QueryConfig section.

func WithReference added in v0.0.25 

func WithReference(r ReferenceConfig) Option

WithReference replaces the ReferenceConfig section.

func WithSchemaRegistry added in v0.0.25 

func WithSchemaRegistry(sr SchemaRegistry) Option

WithSchemaRegistry sets the schema registry on the config.

func WithTransaction added in v0.0.25 

func WithTransaction(t TransactionConfig) Option

WithTransaction replaces the TransactionConfig section.

type OrderBy 

type OrderBy struct {
	Attribute string    `json:"attribute"`
	SortOrder SortOrder `json:"sort_order,omitempty"`
}

type PerformanceConfig added in v0.0.2 

type PerformanceConfig struct {
	EnableMonitoring          bool          `json:"enableMonitoring"`
	SlowQueryThreshold        time.Duration `json:"slowQueryThreshold"`
	SlowOperationThreshold    time.Duration `json:"slowOperationThreshold"`
	MetricsCollectionInterval time.Duration `json:"metricsCollectionInterval"`
	BatchSize                 int           `json:"batchSize"`
	MaxBatchSize              int           `json:"maxBatchSize"`
	Batch                     BatchConfig   `json:"batch"`

	// Unified monitoring settings
	MaxMetricsHistory      int           `json:"maxMetricsHistory"`
	MaxAlertsHistory       int           `json:"maxAlertsHistory"`
	MaxRecommendations     int           `json:"maxRecommendations"`
	EnableAlerting         bool          `json:"enableAlerting"`
	EnableRecommendations  bool          `json:"enableRecommendations"`
	AlertingInterval       time.Duration `json:"alertingInterval"`
	RecommendationInterval time.Duration `json:"recommendationInterval"`

	// Memory monitoring
	EnableMemoryMonitoring bool  `json:"enableMemoryMonitoring"`
	MemoryThreshold        int64 `json:"memoryThreshold"`

	// Correlation tracking
	EnableCorrelationTracking bool          `json:"enableCorrelationTracking"`
	CorrelationTTL            time.Duration `json:"correlationTTL"`
}

PerformanceConfig contains performance monitoring settings

type PropertySchema added in v0.0.15 

type PropertySchema struct {
	Name       string                     `json:"name"`
	Type       string                     `json:"type"` // "string", "integer", "number", "boolean", "array", "object", "null"
	Format     string                     `json:"format,omitempty"`
	Items      *PropertySchema            `json:"items,omitempty"`
	Properties map[string]*PropertySchema `json:"properties,omitempty"`
	Required   bool                       `json:"required"`
	Default    any                        `json:"default,omitempty"`
	Enum       []any                      `json:"enum,omitempty"`
	Minimum    *float64                   `json:"minimum,omitempty"`
	Maximum    *float64                   `json:"maximum,omitempty"`
	MinLength  *int                       `json:"minLength,omitempty"`
	MaxLength  *int                       `json:"maxLength,omitempty"`
	Pattern    string                     `json:"pattern,omitempty"`
	Relation   *RelationSchema            `json:"x-relation,omitempty"`
	LTBaseType string                     `json:"x-ltbase-type,omitempty"`      // "virtual" for virtual fields that are populated dynamically
	LTBaseNote string                     `json:"x-ltbase-note-prop,omitempty"` // Reference to note field: "${note_id}", "${owner_id}", "${note_data}"
}

PropertySchema defines the schema for a single property.

type QueryConfig added in v0.0.2 

type QueryConfig struct {
	DefaultTimeout     time.Duration `json:"defaultTimeout"`
	MaxRows            int           `json:"maxRows"`
	DefaultPageSize    int           `json:"defaultPageSize"`
	MaxPageSize        int           `json:"maxPageSize"`
	EnableQueryPlan    bool          `json:"enableQueryPlan"`
	EnableOptimization bool          `json:"enableOptimization"`
	CacheQueryPlans    bool          `json:"cacheQueryPlans"`
	QueryPlanCacheTTL  time.Duration `json:"queryPlanCacheTTL"`
}

QueryConfig contains query execution settings

type QueryRequest 

type QueryRequest struct {
	SchemaName   string                 `json:"schema_name" validate:"required"`
	Page         int                    `json:"page" validate:"min=1"`
	ItemsPerPage int                    `json:"items_per_page" validate:"min=1,max=100"`
	Condition    Condition              `json:"-"` // Custom unmarshal, can be CompositeCondition or KvCondition
	SortBy       []string               `json:"sort_by,omitempty"`
	SortOrder    SortOrder              `json:"sort_order,omitempty"`
	RowID        *uuid.UUID             `json:"row_id,omitempty"` // For entity-specific operations
	Attrs        []string               `json:"attrs,omitempty"`  // Attributes to return (field projection)
	Federated    *FederatedQueryRequest `json:"federated,omitempty"`
}

QueryRequest represents a pagination query request.

func (QueryRequest) MarshalJSON 

func (r QueryRequest) MarshalJSON() ([]byte, error)

MarshalJSON implements custom JSON marshaling for QueryRequest.

func (*QueryRequest) UnmarshalJSON 

func (r *QueryRequest) UnmarshalJSON(data []byte) error

UnmarshalJSON implements custom JSON unmarshaling for QueryRequest. It allows the Condition field to be either a CompositeCondition or KvCondition.

type QueryResult 

type QueryResult struct {
	Data          []*DataRecord `json:"data"`
	TotalRecords  int           `json:"total_records"`
	TotalPages    int           `json:"total_pages"`
	CurrentPage   int           `json:"current_page"`
	ItemsPerPage  int           `json:"items_per_page"`
	HasNext       bool          `json:"has_next"`
	HasPrevious   bool          `json:"has_previous"`
	ExecutionTime time.Duration `json:"execution_time"`
}

QueryResult represents paginated query results.

type Reference 

type Reference struct {
	SourceSchemaName string        `json:"sourceSchemaName"`
	SourceRowID      uuid.UUID     `json:"sourceRowId"`
	SourceFieldName  string        `json:"sourceFieldName"`
	TargetSchemaName string        `json:"targetSchemaName"`
	TargetRowID      uuid.UUID     `json:"targetRowId"`
	ReferenceType    ReferenceType `json:"referenceType"`
}

Reference represents a reference from one entity to another

type ReferenceConfig added in v0.0.2 

type ReferenceConfig struct {
	ValidateOnCreate bool                   `json:"validateOnCreate"`
	ValidateOnUpdate bool                   `json:"validateOnUpdate"`
	CheckIntegrity   bool                   `json:"checkIntegrity"`
	CascadeDelete    bool                   `json:"cascadeDelete"`
	CascadeUpdate    bool                   `json:"cascadeUpdate"`
	MaxCascadeDepth  int                    `json:"maxCascadeDepth"`
	CascadeRules     map[string]CascadeRule `json:"cascadeRules,omitempty"`
	EnableCaching    bool                   `json:"enableCaching"`
	CacheTTL         time.Duration          `json:"cacheTTL"`
	MaxCacheSize     int                    `json:"maxCacheSize"`
	BatchSize        int                    `json:"batchSize"`
}

ReferenceConfig contains reference management settings

type ReferenceType 

type ReferenceType string

ReferenceType represents the type of reference 

const (
	ReferenceTypeSingle ReferenceType = "single"
	ReferenceTypeArray  ReferenceType = "array"
	ReferenceTypeNested ReferenceType = "nested"
)

type RelationSchema added in v0.0.15 

type RelationSchema struct {
	Target      string `json:"target"`       // Target schema name
	Type        string `json:"type"`         // "reference" for foreign key relationships
	KeyProperty string `json:"key_property"` // child-side foreign key attribute
}

RelationSchema defines reference relationships between objects.

type RequiredPolicy added in v0.0.27 

type RequiredPolicy string

RequiredPolicy defines when an attribute is required. 

const (
	RequiredPolicyOptional        RequiredPolicy = "optional"
	RequiredPolicyAlways          RequiredPolicy = "required_always"
	RequiredPolicyIfParentPresent RequiredPolicy = "required_if_parent_present"
)

type RoutingPolicy added in v0.0.23 

type RoutingPolicy struct {
	Strategy          RoutingStrategy `json:"strategy"`          // "freshness-first", "cost-first", "hybrid"
	HotTTL            time.Duration   `json:"hotTTL"`            // TTL to consider data "hot"
	MaxDuckDBScanRows int             `json:"maxDuckDBScanRows"` // threshold for preferring cold scans
	AllowS3Fallback   bool            `json:"allowS3Fallback"`   // allow falling back to S3/DuckDB when PG not used
}

RoutingPolicy defines federated query routing behavior

type RoutingStrategy added in v0.0.25 

type RoutingStrategy string

RoutingStrategy specifies the federated query routing algorithm. 

const (
	// RoutingStrategyFreshnessFirst prefers the hot (PostgreSQL) tier for
	// queries that explicitly request fresh data (PreferHot flag).
	RoutingStrategyFreshnessFirst RoutingStrategy = "freshness-first"

	// RoutingStrategyCostFirst routes large scans to DuckDB to reduce
	// PostgreSQL load; small scans stay on the hot tier.
	RoutingStrategyCostFirst RoutingStrategy = "cost-first"

	// RoutingStrategyHybrid uses DuckDB by default but short-circuits to
	// PostgreSQL when the result set is expected to be small or hot data is
	// explicitly preferred.
	RoutingStrategyHybrid RoutingStrategy = "hybrid"
)

type SchemaAttributeCache added in v0.0.3 

type SchemaAttributeCache map[string]AttributeMetadata

SchemaAttributeCache is a mapping of attr_name -> metadata. Strongly recommended to populate per schema_id at application startup.

type SchemaRegistry added in v0.0.3 

type SchemaRegistry interface {
	// GetSchemaAttributeCacheByName retrieves schema ID and attribute cache by schema name
	GetSchemaAttributeCacheByName(name string) (int16, SchemaAttributeCache, error)
	// GetSchemaAttributeCacheByID retrieves schema name and attribute cache by schema ID
	GetSchemaAttributeCacheByID(id int16) (string, SchemaAttributeCache, error)

	GetSchemaByName(name string) (int16, JSONSchema, error)
	// GetSchemaAttributeCacheByID retrieves schema name and attribute cache by schema ID
	GetSchemaByID(id int16) (string, JSONSchema, error)
	ListSchemas() []string
}

SchemaRegistry provides schema lookup operations. Implementations can load schemas from files, databases, or other sources.

type SortOrder 

type SortOrder string

SortOrder defines sort direction 

const (
	SortOrderAsc  SortOrder = "asc"
	SortOrderDesc SortOrder = "desc"
)

type TableNames added in v0.0.2 

type TableNames struct {
	SchemaRegistry string `json:"schemaRegistry"`
	EntityMain     string `json:"entityMain"`
	EAVData        string `json:"eavData"`
	ChangeLog      string `json:"changeLog"`
}

TableNames generates the table names for a specific client and project

type TransactionConfig added in v0.0.2 

type TransactionConfig struct {
	DefaultTimeout           time.Duration `json:"defaultTimeout"`
	MaxTimeout               time.Duration `json:"maxTimeout"`
	MaxRetryAttempts         int           `json:"maxRetryAttempts"`
	RetryAttempts            int           `json:"retryAttempts"`
	RetryDelay               time.Duration `json:"retryDelay"`
	IsolationLevel           string        `json:"isolationLevel"`
	EnableDeadlockDetection  bool          `json:"enableDeadlockDetection"`
	DeadlockCheckInterval    time.Duration `json:"deadlockCheckInterval"`
	DeadlockMaxWaitTime      time.Duration `json:"deadlockMaxWaitTime"`
	SlowTransactionThreshold time.Duration `json:"slowTransactionThreshold"`
	MinSuccessRate           float64       `json:"minSuccessRate"`
	MaxAverageDuration       time.Duration `json:"maxAverageDuration"`
	MaxConnectionPoolUsage   float64       `json:"maxConnectionPoolUsage"`
}

TransactionConfig contains transaction settings

type ValueType added in v0.0.3 

type ValueType string

ValueType represents supported attribute value types. 

const (
	ValueTypeText     ValueType = "text"
	ValueTypeSmallInt ValueType = "smallint"
	ValueTypeInteger  ValueType = "integer"
	ValueTypeBigInt   ValueType = "bigint"
	ValueTypeNumeric  ValueType = "numeric"  // double precision
	ValueTypeDate     ValueType = "date"     // for JSON attributes with format `date`
	ValueTypeDateTime ValueType = "datetime" // for JSON attributes with format `date-time`
	ValueTypeUUID     ValueType = "uuid"
	ValueTypeBool     ValueType = "bool"
	ValueTypeList     ValueType = "list" // array type stored as DuckDB LIST in parquet
)

Source Files

View all Source files

Directories

Path Synopsis
cmd
benchmark command
lambda command
Package main provides the AWS Lambda entry point for the Forma API server.
 Package main provides the AWS Lambda entry point for the Forma API server.
sample command
server command
tools command
bootstrap
cdc
compaction
conditionexpr
e2e_harness
e2e_harness/federated
Package federated provides custom assertions for E2E testing.
 Package federated provides custom assertions for E2E testing.
e2e_harness/federated/benchmark
httpapi
manifest
queryoptimizer
telemetry

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.