forge

package
v0.10.0 Latest Latest
Warning

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

 Go to latest Published: Jun 12, 2026 License: MIT Imports: 10 Imported by: 0

Documentation

Overview

Package forge provides governed, self-documenting KPI computation functions.

A Function[In, Out] carries its input and output codecs, a SHA-256 contract hash, version string, and optional governance metadata (author, approver, approval date) — all in one value. The contract hash changes whenever the codec or the compute function changes, making silent drift impossible.

Defining a function 

var oeeFunction = forge.NewFunction[OEEInput, OEEResult](
    "oee", "1.0.0",
    oeeInputCodec, oeeResultCodec,
    func(ctx context.Context, in OEEInput) (OEEResult, error) {
        return OEEResult{OEE: in.Availability * in.Performance * in.Quality}, nil
    },
    forge.FunctionMeta{
        Description: "Overall Equipment Effectiveness (ISO 22400-2).",
        Author:      "engineering@example.com",
        ApprovedBy:  "quality-board",
        ApprovedAt:  "2024-01-15",
    },
)

Validation sequence

When Function.Apply is called:

  1. Input codec decodes and validates → InputError on failure
  2. Optional cross-input refinement runs → RefinementError on failure
  3. User function executes → ApplyError on failure
  4. Output codec validates → OutputError on failure

Composing functions

Compose chains two functions type-safely: 

var pipeline = forge.Compose(rawDataFn, oeeFn)
// type-checked: Out of rawDataFn must match In of oeeFn

Registering in a pipeline 

registry := forge.NewRegistry("OEE Pipeline", "1.0.0").
    WithAuthor("engineering@example.com").
    WithApproval("quality-board", "2024-01-15").
    WithObserver(obs)

registry.Register(oeeFunction)
spec := registry.PipelineSpec()
// render with render/pipeline to produce a YAML governance document

Collection operations

Lift a function over a slice or map: 

batchOEE  := forge.Map(oeeFunction)    // []OEEInput → []OEEResult
alerts    := forge.Filter(alertFn)     // []Reading → []Reading
total     := forge.Reduce(sumFn)       // ([]Reading, float64) → float64

Further reading

Package forge provides signed, governed, self-documenting KPI computation functions for the go-codex library.

Forge adds a third composable layer on top of validated domain models (codex) and API builders (api/rest, api/events): 

Layer 1: Validated domain models — codex.Codec[T] with Refine constraints.
Layer 2: API endpoints        — api/rest or api/events builders.
Layer 3: KPI pipelines        — forge.Function[In,Out] with governance + computation graph.

The layers are independent. Use any subset; they compose naturally.

Measured[T]: boundary provenance

Within a computation pipeline, values are naked validated types (e.g. Availability, Performance). When a value crosses a system boundary (REST request, MQTT message, database read), wrap it with Measured[T] to carry governance provenance: 

codec := forge.MeasuredCodec(availabilityCodec)
m := forge.Measured[Availability]{
    Source: "sensor-ot-1", Version: "2.0", Author: "OT Team",
    Value:  av,
}
encoded, _ := codec.Encode(m)  // wire form includes source/version/author

Functions: signed, validated computations

Function[In, Out] wraps a Go computation with a validated input and output, plus governance metadata (Author, ApprovedBy, ApprovedAt) and a tamper-evident hash of the computation contract (name + version + input/output schemas).

For single-value inputs, In is a plain domain type. Port names for pipeline graph-edge inference come from codec.Schema.Title (set via .WithTitle): 

var oeeCodec   = codex.Float64(zeroToOne()).WithTitle("oee")
var gradeCodec = codex.String(gradeEnum).WithTitle("grade")

gradeCalc := forge.NewFunction("gradeCalc", "1.0.0",
    oeeCodec, gradeCodec,
    func(oee OEE) (Grade, error) { ... },
)

For multi-input computations, define a struct and build a codex.Struct codec. Struct field names are used as input port names automatically (Schema.Title on the struct codec is not needed for port naming). Cross-field constraints belong on the struct codec via codex.Refine; use WithRefinement for pipeline-level constraints: 

type OEEIn struct {
    Availability Availability
    Performance  Performance
    Quality      Quality
}
oeeInCodec := codex.Struct[OEEIn](
    codex.RequiredField("availability", availabilityCodec, ...),
    codex.RequiredField("performance",  performanceCodec, ...),
    codex.RequiredField("quality",      qualityCodec, ...),
)
var oeeCodec = codex.Float64(zeroToOne()).WithTitle("oee")

oeeCalc := forge.NewFunction("oeeCalc", "1.0.0",
    oeeInCodec, oeeCodec,
    func(in OEEIn) (OEE, error) {
        return OEE(float64(in.Availability) * float64(in.Performance) * float64(in.Quality)), nil
    },
    forge.FunctionMeta{Description: "Computes OEE as availability × performance × quality.", Author: "OT Engineering", ApprovedBy: "Quality Manager", ApprovedAt: "2024-03-01"},
)
result, err := oeeCalc.Apply(OEEIn{Availability: av, Performance: pe, Quality: qu})

Registry + spec: machine-readable computation graph

Register functions and call Spec() to produce a PipelineSpec that render/pipeline serialises as YAML — a machine-readable, git-committable audit trail of the entire KPI computation graph: 

reg := forge.NewRegistry("OEE Pipeline", "1.0.0").
    WithDescription("Signed, governed OEE computation pipeline.")
oeeCalc.Register(reg)
yamlBytes, _ := pipeline.Render(reg.Spec())

Index

Examples

Constants

This section is empty.

Variables

This section is empty.

Functions

func MeasuredCodec 

func MeasuredCodec[T any](inner codex.Codec[T]) codex.Codec[Measured[T]]

MeasuredCodec wraps an existing Codec[T] and produces a Codec[Measured[T]].

The encoded form carries "source", "version", "author", and "value" fields. Source, Version, and Author are validated as non-empty strings on both encode and decode. The inner codec's constraints apply to Value on both encode and decode.

Types

type ApplyError 

type ApplyError struct {
	Function string
	Err      error
}

ApplyError is returned by Function*.Apply when the compute function itself returns an error. Function is the function's name, Err is the error returned by the compute function.

func (ApplyError) Error 

func (e ApplyError) Error() string

func (ApplyError) LogValue 

func (e ApplyError) LogValue() slog.Value

LogValue implements slog.LogValuer for structured logging.

func (ApplyError) Unwrap 

func (e ApplyError) Unwrap() error

type CollectionElementError 

type CollectionElementError struct {
	Function string
	Index    int
	Err      error
}

CollectionElementError is returned by Map, Filter, or Reduce when a collection operation fails at a specific element index. Function is the collection function's name, Index is the 0-based element position, Err is the underlying error.

func (CollectionElementError) Error 

func (e CollectionElementError) Error() string

func (CollectionElementError) LogValue 

func (e CollectionElementError) LogValue() slog.Value

LogValue implements slog.LogValuer for structured logging.

func (CollectionElementError) Unwrap 

func (e CollectionElementError) Unwrap() error

type CollectionKeyError 

type CollectionKeyError struct {
	Function string
	Key      string
	Err      error
}

CollectionKeyError is returned by MapValues when an operation fails at a specific map key. Function is the collection function's name, Key is the failing map key, Err is the underlying error.

func (CollectionKeyError) Error 

func (e CollectionKeyError) Error() string

func (CollectionKeyError) LogValue 

func (e CollectionKeyError) LogValue() slog.Value

LogValue implements slog.LogValuer for structured logging.

func (CollectionKeyError) Unwrap 

func (e CollectionKeyError) Unwrap() error

type ConfigError 

type ConfigError struct {
	Func  string
	Field string
}

constructor argument is missing. Func identifies the constructor (e.g. "forge.New"), Field is "name" or "version".

func (ConfigError) Error 

func (e ConfigError) Error() string

func (ConfigError) LogValue 

func (e ConfigError) LogValue() slog.Value

LogValue implements slog.LogValuer for structured logging.

type Function 

type Function[In, Out any] struct {
	// Spec is the schema-level descriptor: governance metadata + contract hash.
	Spec FunctionSpec
	// contains filtered or unexported fields
}

Function is a validated, signed derivation function with a single generic input In and output Out.

For single-value inputs (e.g. computing a grade from an OEE value), In is a plain domain type. For multi-input computations, define a struct that groups the inputs and build its codec with codex.Struct — each field is validated individually by the struct codec, and cross-field constraints can be added via codex.Refine on the struct codec or via WithRefinement at the forge.New call site.

Apply runs the following sequence:

  1. Validate In with the input codec → InputError on failure
  2. Run optional WithRefinement constraint → RefinementError on failure
  3. Run the compute function → ApplyError on failure
  4. Validate Out with the output codec → OutputError on failure
  5. Notify the observer

func Compose 

func Compose[A, B, Out any](
	name, version string,
	f1 *Function[A, B],
	f2 *Function[B, Out],
	opts ...FunctionOpt,
) *Function[A, Out]

Compose chains two Functions: the output of f1 becomes the input of f2.

The resulting Function validates in through f1's input codec, runs f1 (including any refinement on f1), then feeds the result into f2 (including any refinement on f2). The composed FunctionSpec records its own contract hash from name, version, and the outer input/output shapes.

Pass WithRefinement to add a pipeline-level cross-input constraint on the composed function's input (type A). It runs after f1's input codec validation and before f1.

Compose panics if name or version is empty — these are programming errors. 

celsius2kelvin := forge.Compose("c2k", "1.0.0", celsius2centi, centi2kelvin)

func Filter 

func Filter[T any](
	name, version string,
	elemCodec codex.Codec[T],
	pred func(T) bool,
	opts ...FunctionOpt,
) *Function[[]T, []T]

Filter returns a Function[[]T, []T] that keeps only elements satisfying pred.

Each element is validated by elemCodec before pred is called. On the first element validation failure, Filter returns a CollectionElementError. Elements for which pred returns false are silently dropped (no error).

The FunctionSpec carries Kind="filter". Panics if name or version is empty — these are programming errors. 

warmFilter := forge.Filter("warmFilter", "1.0.0", rawCodec,
    func(r RawReading) bool { return r.WarmUp },
)

func Map 

func Map[In, Out any](
	name, version string,
	fn *Function[In, Out],
	opts ...FunctionOpt,
) *Function[[]In, []Out]

Map lifts fn to operate element-wise over a slice: []In → []Out.

Each element is validated and processed by fn. On the first element failure (input validation, refinement, compute, or output validation), Map returns a CollectionElementError carrying the element index and the underlying error.

The resulting Function carries Kind="map" and Wraps=fn.Spec.Name in its FunctionSpec, making the relationship visible in the pipeline YAML spec.

Governance options (FunctionMeta and WithRefinement) work on the slice-level function. WithRefinement receives the whole []In slice — use it for collection-level constraints (e.g. minimum count).

Panics if name or version is empty — these are programming errors. 

batchCalc := forge.Map("batchCalc", "1.0.0", oeeCalc,
    forge.FunctionMeta{Description: "Applies oeeCalc to a batch of measurements."},
    forge.WithRefinement(func(in []OEEIn) error {
        if len(in) == 0 {
            return fmt.Errorf("batch must not be empty")
        }
        return nil
    }),
)

func MapValues 

func MapValues[In, Out any](
	name, version string,
	fn *Function[In, Out],
	opts ...FunctionOpt,
) *Function[map[string]In, map[string]Out]

MapValues lifts fn to operate over all values of a map[string]In → map[string]Out.

Each value is processed by fn. On the first key failure (input validation, refinement, compute, or output validation), MapValues returns a CollectionKeyError carrying the key and the underlying error. Iteration order is not guaranteed.

The FunctionSpec carries Kind="mapValues" and Wraps=fn.Spec.Name.

Map keys are not validated by MapValues. To enforce a key format, use MapValuesK. Panics if name or version is empty — these are programming errors. 

batchPerSensor := forge.MapValues("batchPerSensor", "1.0.0", summariseCalc)

func MapValuesK 

func MapValuesK[K comparable, In, Out any](
	name, version string,
	keyCodec codex.Codec[K],
	fn *Function[In, Out],
	opts ...FunctionOpt,
) *Function[map[K]In, map[K]Out]

MapValuesK lifts fn to operate over all values of a map[K]In → map[K]Out, with key validation enforced by keyCodec before any element is processed.

Key validation is fail-fast: the first key that fails keyCodec causes Apply to return InputError → KeyError → ConstraintError. No values are processed until all keys pass. This mirrors the behaviour of codex.Map[K, V] used as an input codec.

Per-value failures (input, refinement, compute, or output within fn) are returned as CollectionKeyError, same as MapValues.

K must satisfy comparable and keyCodec must encode to string (JSON/YAML requirement). Use codex.String().Refine(...) for pattern or enum constraints on string keys.

The FunctionSpec carries Kind="mapValues" and Wraps=fn.Spec.Name. The input schema gains a "propertyNames" entry from keyCodec.Schema.

Panics if name or version is empty — these are programming errors. 

perSensor := forge.MapValuesK("perSensor", "1.0.0",
    sensorIDCodec,  // enforces ^[a-z]+-\d+$ on every key
    singleSensorPipeline,
    forge.FunctionMeta{Description: "Applies the sensor pipeline to every validated sensor key."},
)

func NewFunction 

func NewFunction[In, Out any](
	name, version string,
	input codex.Codec[In],
	output codex.Codec[Out],
	apply func(In) (Out, error),
	opts ...FunctionOpt,
) *Function[In, Out]

NewFunction creates a Function and computes its contract hash. Panics if name or version is empty — these are programming errors detected at startup, not runtime conditions.

NewFunction is a free function (not a method) because Go requires type parameters to appear on free functions, not on method receivers.

Port names (used for pipeline graph-edge inference) are inferred from the codec's Schema.Title (set via codex.Codec.WithTitle). For struct codecs the struct field names are always used regardless of Title. Scalar codecs default to "input"/"output" when no title is set.

For single-value inputs pass any scalar codec directly: 

var oeeCodec   = codex.Float64(zeroToOne()).WithTitle("oee")
var gradeCodec = codex.String(gradeEnum).WithTitle("grade")

gradeCalc := forge.NewFunction("gradeCalc", "1.0.0",
    oeeCodec, gradeCodec,
    func(oee OEE) (Grade, error) { ... },
)

For multi-input computations, define an input struct and a codex.Struct codec. Cross-field constraints belong on the struct codec via codex.Refine; use WithRefinement for pipeline-level constraints: 

type AvailabilityIn struct {
    PlannedTime PlannedTime
    Downtime    Downtime
}
availInCodec := codex.Struct[AvailabilityIn](
    codex.RequiredField("plannedTime", ptCodec, ...),
    codex.RequiredField("downtime",    dtCodec, ...),
)
availabilityCodec := codex.Float64(zeroToOne()).WithTitle("availability")

availCalc := forge.NewFunction("availabilityCalc", "1.0.0",
    availInCodec, availabilityCodec,
    func(in AvailabilityIn) (Availability, error) { ... },
    forge.WithRefinement(func(in AvailabilityIn) error { ... }),
    forge.FunctionMeta{Description: "Computes availability as (plannedTime - downtime) / plannedTime."},
)
Example 
package main

import (
	"fmt"

	"github.com/DaniDeer/go-codex/codex"
	"github.com/DaniDeer/go-codex/forge"
)

func main() {
	type OEEInput struct {
		Availability float64
		Performance  float64
		Quality      float64
	}
	type OEEResult struct {
		OEE float64
	}

	inputCodec := codex.Struct[OEEInput](
		codex.RequiredField("availability", codex.Float64(),
			func(v OEEInput) float64 { return v.Availability },
			func(v *OEEInput, f float64) { v.Availability = f },
		),
		codex.RequiredField("performance", codex.Float64(),
			func(v OEEInput) float64 { return v.Performance },
			func(v *OEEInput, f float64) { v.Performance = f },
		),
		codex.RequiredField("quality", codex.Float64(),
			func(v OEEInput) float64 { return v.Quality },
			func(v *OEEInput, f float64) { v.Quality = f },
		),
	)
	outputCodec := codex.Struct[OEEResult](
		codex.RequiredField("oee", codex.Float64(),
			func(v OEEResult) float64 { return v.OEE },
			func(v *OEEResult, f float64) { v.OEE = f },
		),
	)

	// NewFunction creates a governed, self-documenting computation.
	fn := forge.NewFunction[OEEInput, OEEResult](
		"oee", "1.0.0",
		inputCodec, outputCodec,
		func(in OEEInput) (OEEResult, error) {
			return OEEResult{OEE: in.Availability * in.Performance * in.Quality}, nil
		},
		forge.FunctionMeta{Description: "Overall Equipment Effectiveness"},
	)

	result, err := fn.Apply(OEEInput{
		Availability: 0.9, Performance: 0.85, Quality: 0.95,
	})
	if err != nil {
		fmt.Println("error:", err)
		return
	}
	fmt.Printf("OEE=%.4f\n", result.OEE)
}

Output:
OEE=0.7268

func Reduce 

func Reduce[T, Acc any](
	name, version string,
	elemCodec codex.Codec[T],
	accCodec codex.Codec[Acc],
	init Acc,
	step func(Acc, T) Acc,
	opts ...FunctionOpt,
) *Function[[]T, Acc]

Reduce returns a Function[[]T, Acc] that folds the slice to a single accumulator.

Each element is validated by elemCodec before the step function is called. On the first element validation failure, Reduce returns a CollectionElementError. The final accumulator is validated by accCodec before being returned.

The FunctionSpec carries Kind="reduce". Panics if name or version is empty — these are programming errors. 

summarise := forge.Reduce("summarise", "1.0.0",
    celsiusCodec, summaryCodec,
    BatchSummary{},
    func(acc BatchSummary, c Celsius) BatchSummary { ... },
)

func (*Function[In, Out]) Apply 

func (f *Function[In, Out]) Apply(in In) (Out, error)

Apply validates in, runs the optional cross-input refinement, runs the computation, then validates and returns the result.

func (*Function[In, Out]) Register 

func (f *Function[In, Out]) Register(r *Registry) *Registry

Register adds this function to r, injects r's observer, and returns r for chaining.

type FunctionKind 

type FunctionKind string

FunctionKind identifies how a Function was constructed. 

const (
	// FunctionKindScalar is the default: a scalar function created by NewFunction or Compose.
	// Its value is "" because scalar is the implicit kind — NewFunction and Compose
	// do not write Kind explicitly, so the zero value is the correct sentinel.
	FunctionKindScalar FunctionKind = ""
	// FunctionKindMap is a function created by Map (lifts a Function over a slice).
	FunctionKindMap FunctionKind = "map"
	// FunctionKindFilter is a function created by Filter (keeps elements satisfying a predicate).
	FunctionKindFilter FunctionKind = "filter"
	// FunctionKindReduce is a function created by Reduce (folds a slice to an accumulator).
	FunctionKindReduce FunctionKind = "reduce"
	// FunctionKindMapValues is a function created by MapValues or MapValuesK.
	FunctionKindMapValues FunctionKind = "mapValues"
)

type FunctionMeta 

type FunctionMeta struct {
	// Description is a short human-readable explanation of what the function computes.
	Description string
	// Author is the team or person responsible for this function definition.
	Author string
	// ApprovedBy names the approver or approval authority.
	ApprovedBy string
	// ApprovedAt is the ISO 8601 date of approval (e.g. "2024-03-01").
	ApprovedAt string
}

FunctionMeta holds optional governance metadata for a forge function. It implements FunctionOpt and can be passed directly to NewFunction or Compose. 

forge.NewFunction("calc", "1.0.0", inCodec, outCodec, fn,
    forge.FunctionMeta{
        Description: "compute efficiency grade",
        Author:      "OT Engineering",
        ApprovedBy:  "Quality Manager",
        ApprovedAt:  "2024-03-01",
    },
)

type FunctionOpt 

type FunctionOpt interface {
	// contains filtered or unexported methods
}

FunctionOpt configures optional governance metadata or cross-input validation on a forge function.

Pass one or more FunctionOpt values as trailing variadic arguments to NewFunction or Compose. The two implementations are:

  • FunctionMeta struct literal — for governance metadata (description, author, approval)
  • WithRefinement — for a typed pipeline-level cross-input constraint

Governance fields not supplied default to the zero string ("") and are omitted from the YAML spec output.

func WithRefinement 

func WithRefinement[In any](fn func(In) error) FunctionOpt

WithRefinement adds a pipeline-level cross-input constraint to a function.

The constraint runs after input codec validation and before the compute function. Return a non-nil error to reject the input. On failure, Apply returns a RefinementError.

For multi-input functions the constraint receives the assembled input struct, giving access to all fields. Cross-field rules can also be expressed directly on the input codec via codex.Refine — prefer that approach when the constraint is a property of the domain type itself rather than the pipeline definition. 

availCalc := forge.NewFunction("availabilityCalc", "1.0.0",
    availInCodec, availabilityCodec,
    computeAvailability,
    forge.WithRefinement(func(in AvailabilityIn) error {
        if float64(in.Downtime) > float64(in.PlannedTime) {
            return fmt.Errorf("downtime (%v h) must not exceed plannedTime (%v h)",
                in.Downtime, in.PlannedTime)
        }
        return nil
    }),
)

type FunctionSpec 

type FunctionSpec struct {
	// Core identity — always set.
	Name    string
	Version string
	// Hash is "sha256:<hex>" over the canonical JSON of the computation contract.
	Hash string
	// Kind identifies the constructor that produced this function.
	// FunctionKindScalar (empty string) for scalar functions created by NewFunction or Compose.
	// FunctionKindMap/Filter/Reduce/MapValues for collection functions.
	Kind FunctionKind
	// Wraps is the Name of the scalar Function lifted by Map or MapValues.
	// Empty for scalar functions and for Filter/Reduce (which take raw predicates).
	Wraps string
	// Governance metadata — set via [FunctionMeta].
	Description string
	Author      string
	ApprovedBy  string
	ApprovedAt  string // ISO 8601 date, e.g. "2024-01-15"
	// Input/output shapes.
	Inputs []PortSpec
	Output PortSpec
}

FunctionSpec is the type-erased, schema-level descriptor of a function.

All fields are set automatically by NewFunction, Compose, and the collection constructors (Map, Filter, Reduce, MapValues). Hash is computed over (Name, Version, Inputs[].Schema, Output.Schema); governance fields (Author, ApprovedBy, ApprovedAt) are excluded from the hash — changing who approved a function does not alter what it computes.

type GraphEdge 

type GraphEdge struct {
	Function   string // name of the consuming function
	Input      string // input name within that function
	ProducedBy string // name of the producing function, or "" for direct measurements
}

GraphEdge links a consuming function's named input to the function that produces it. ProducedBy is empty when the input is a direct measurement with no registered producer.

type InputError 

type InputError struct {
	Function string
	Input    string
	Err      error
}

InputError is returned by Function*.Apply when an input value fails its codec validation. Function is the function's name, Input is the input parameter's name, Err is the underlying error.

func (InputError) Error 

func (e InputError) Error() string

func (InputError) LogValue 

func (e InputError) LogValue() slog.Value

LogValue implements slog.LogValuer for structured logging.

func (InputError) Unwrap 

func (e InputError) Unwrap() error

type Measured 

type Measured[T any] struct {
	Source  string `json:"source"`
	Version string `json:"version"`
	Author  string `json:"author"`
	Value   T      `json:"value"`
}

Measured[T] wraps a validated value with its governance provenance.

Use Measured[T] when a validated value must carry its data lineage across a system boundary (REST API, MQTT message, database row). Within a computation pipeline, pass the unwrapped Value field directly to Function*.Apply — no wrapper overhead at the computation layer.

type OutputError 

type OutputError struct {
	Function string
	Output   string
	Err      error
}

OutputError is returned by Function*.Apply when the computed output fails its codec validation. Function is the function's name, Output is the output parameter's name, Err is the underlying error.

func (OutputError) Error 

func (e OutputError) Error() string

func (OutputError) LogValue 

func (e OutputError) LogValue() slog.Value

LogValue implements slog.LogValuer for structured logging.

func (OutputError) Unwrap 

func (e OutputError) Unwrap() error

type PipelineInfo 

type PipelineInfo struct {
	Title       string
	Version     string
	Description string
	// Author is the team or person responsible for this pipeline definition.
	// Mirrors the per-function [FunctionMeta.Author] field at the pipeline level.
	Author string
	// ApprovedBy names the approval authority for the pipeline as a whole.
	ApprovedBy string
	// ApprovedAt is the ISO 8601 date of pipeline approval (e.g. "2024-03-01").
	ApprovedAt string
}

PipelineInfo is pipeline-level metadata.

type PipelineSpec 

type PipelineSpec struct {
	Info      PipelineInfo
	Functions []FunctionSpec
	Graph     []GraphEdge
}

PipelineSpec is the full machine-readable computation graph spec. Use render/pipeline.Render to serialise it as YAML.

type PortSpec 

type PortSpec struct {
	Name   string        `json:"name"`
	Schema schema.Schema `json:"schema"`
}

PortSpec describes one named input or output port of a function. Schema is the codec's schema for this value.

type RefinementError 

type RefinementError struct {
	Function string
	Err      error
}

RefinementError is returned by Function*.Apply when a cross-input constraint fails. It is produced after all individual input validations pass but before the compute function runs. Function is the function's name, Err is the constraint error returned by the refinement function.

func (RefinementError) Error 

func (e RefinementError) Error() string

func (RefinementError) LogValue 

func (e RefinementError) LogValue() slog.Value

LogValue implements slog.LogValuer for structured logging.

func (RefinementError) Unwrap 

func (e RefinementError) Unwrap() error

type Registry 

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

Registry holds registered functions and produces a PipelineSpec.

Graph edges are inferred automatically: when function A's output name matches function B's input name, the registry records that B depends on A.

Build a registry with NewRegistry and chain optional configuration via WithDescription, WithAuthor, WithApproval, and WithObserver before registering any functions: 

reg := forge.NewRegistry("OEE Pipeline", "1.0.0").
    WithDescription("Signed, governed OEE computation pipeline.").
    WithAuthor("OT Engineering").
    WithApproval("Quality Manager", "2024-03-01").
    WithObserver(myObs)
availabilityCalc.Register(reg)
oeeCalc.Register(reg)

func NewRegistry 

func NewRegistry(title, version string) *Registry

NewRegistry returns a new Registry with the given pipeline title and version. Chain WithDescription and WithObserver to add optional configuration.

func (*Registry) Spec 

func (r *Registry) Spec() PipelineSpec

Spec builds and returns the PipelineSpec, including inferred graph edges.

Graph edges are inferred by matching each function's output name against the input names of all other functions. If multiple registered functions share the same output name, all matches are recorded as separate edges.

func (*Registry) WithApproval 

func (r *Registry) WithApproval(approvedBy, approvedAt string) *Registry

WithApproval sets the pipeline-level approver and approval date, returning r for chaining. approvedAt should be an ISO 8601 date string (e.g. "2024-03-01"). Mirrors [FunctionMeta.ApprovedBy] and [FunctionMeta.ApprovedAt] at the pipeline level.

func (*Registry) WithAuthor 

func (r *Registry) WithAuthor(author string) *Registry

WithAuthor sets the pipeline-level author and returns r for chaining. Mirrors [FunctionMeta.Author] at the pipeline level.

func (*Registry) WithDescription 

func (r *Registry) WithDescription(desc string) *Registry

WithDescription sets the pipeline-level description and returns r for chaining.

func (*Registry) WithObserver 

func (r *Registry) WithObserver(obs stats.PipelineObserver) *Registry

WithObserver sets the PipelineObserver injected into every function that registers itself with this registry. Returns r for chaining.

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.