Forge Pipelines¶

// domain/pipeline.go — zero adapter imports
var oeeCalcFn = 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
    },
)

var SensorReadings = codex.Must(ports.NewSourcePort[OEEIn]("sensor-readings", oeeInCodec, ports.PortOptions{}))
var OEEResults = codex.Must(ports.NewSinkPort[OEE]("oee-results", oeeCodec, ports.PortOptions{}))

func StartPipeline(ctx context.Context) {
    sensors := SensorReadings.Stream(ctx)
    oeeStream := gstream.Apply(ctx, sensors, oeeCalcFn, gstream.ApplyOptions{})
    go OEEResults.Feed(ctx, oeeStream)
}

// main.go — all transport decisions here

// Wire A: MQTT source → domain pipeline → MQTT sink
domain.SensorReadings.Bind(ctx, mqtt5.SubscribeAdapter(client, router, sensorHandle, 0, fmt, opts))
domain.OEEResults.Bind(ctx, mqtt5.PublishAdapter(client, alertHandle, fmt, publishOpts))

// Wire B: HTTP trigger → domain pipeline → HTTP response (ToolPort, request/response)
var OEETool = codex.Must(ports.NewToolPort[OEEIn, OEE]("oee-tool", oeeInCodec, oeeCodec, ports.PortOptions{}))
OEETool.SetPipeline(func(ctx context.Context, req OEEIn) gstream.Stream[OEE] {
    return gstream.Apply(ctx, gstream.Single(ctx, req), oeeCalcFn, gstream.ApplyOptions{})
})
domain.OEETool.Bind(ctx, nethttp.PipelineAdapter(mux, httpHandle, nethttp.PipelineAdapterOptions{}))

// Wire C: test — plain Go channel, no transport at all
ch := make(chan OEEIn, 1)
ch <- OEEIn{Availability: 0.9, Performance: 0.85, Quality: 0.95}
close(ch)
domain.SensorReadings.Bind(ctx, ports.ChanSourceAdapter(ch))
out := make(chan OEE, 1)
domain.OEEResults.Bind(ctx, ports.ChanSinkAdapter(out))

// ✅ Correct: pure transformation
func(in OEEIn) (OEE, error) {
    return OEE(float64(in.Availability) * float64(in.Performance) * float64(in.Quality)), nil
}

// ❌ Wrong: I/O inside forge function violates the design
// func(in InputData) (Out, error) {
//     cfg, _ := configFile.Read(...)  ← file I/O
//     resp, _ := nethttp.Call(...)    ← HTTP call
//     row, _ := db.Query(...)         ← database query
//     return combine(in, cfg, resp), nil
// }

// ✅ Correct: external data flows IN as typed input
type EnrichInput struct {
    Sensor SensorReading   // from MQTT stream
    Config ThresholdConfig // from config file stream
}

enrichFn := forge.NewFunction("applyThresholds", "1.0.0",
    enrichInputCodec, alertCodec,
    func(in EnrichInput) (Alert, error) {
        // Pure: both inputs are already validated and available
        if float64(in.Sensor.Value) > in.Config.MaxValue {
            return Alert{Sensor: in.Sensor.ID, Exceeded: true}, nil
        }
        return Alert{}, nil
    },
)

// I/O stays in the stream layer — forge function receives the result:
configs := /* file.WatchStream + FlatMapSlice + ports.File.Read */
combined := stream.CombineLatest2(ctx, sensorStream, configs,
    func(s SensorReading, c ThresholdConfig) EnrichInput { return EnrichInput{s, c} })
alerts := stream.Apply(ctx, combined, enrichFn, stream.ApplyOptions{})

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

// forge.NewFunction is infallible — panics only on empty name or version.
var availabilityCodec = codex.Float64().WithTitle("availability")

availabilityCalc := forge.NewFunction(
    "availabilityCalc", "1.0.0",
    availabilityInCodec,  // Codec[AvailabilityIn] — validates inputs
    availabilityCodec,    // Codec[Availability]   — validates output
    func(in AvailabilityIn) (Availability, error) {
        return Availability(
            (float64(in.PlannedTime) - float64(in.Downtime)) / float64(in.PlannedTime),
        ), nil
    },
    forge.FunctionMeta{
        Description: "Computes availability as (plannedTime - downtime) / plannedTime.",
        Author:      "oee-team",
    },
)

// Apply — input and output are codec-validated; errors are structured.
avail, err := availabilityCalc.Apply(AvailabilityIn{PlannedTime: 8.0, Downtime: 1.0})
var ie forge.InputError
if errors.As(err, &ie) {
    fmt.Printf("input failed: %v\n", ie.Err)
}

type AvailabilityIn struct {
    PlannedTime PlannedTime
    Downtime    Downtime
}

// Cross-field constraint: downtime cannot exceed planned time
var availabilityInCodec = codex.Struct[AvailabilityIn](
    codex.RequiredField("plannedTime", plannedTimeCodec, ...),
    codex.RequiredField("downtime", downtimeCodec, ...),
).RefineFunc(func(a AvailabilityIn) error {
    if float64(a.Downtime) > float64(a.PlannedTime) {
        return fmt.Errorf("downtime exceeds plannedTime")
    }
    return nil
})

forge.NewFunction("calc", "1.0.0", inCodec, outCodec, fn,
    forge.FunctionMeta{
        Description: "Human-readable description",
        Author:      "team-name",
        ApprovedBy:  "reviewer",
        ApprovedAt:  "2024-03-01",
    },
)

// Compose chains f1: A→B and f2: B→Out into Function[A, Out].
// Type-safe: Out of f1 must match In of f2.
combined := forge.Compose("combined", "1.0.0", f1, f2,
    forge.FunctionMeta{Description: "chained pipeline"},
    forge.WithRefinement(func(a A) error { /* pre-compose constraint */ return nil }),
)

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

reg = availabilityCalc.Register(reg)
reg = performanceCalc.Register(reg)
reg = oeeCalc.Register(reg)

// Registry infers graph edges by matching input port names to output port names.
// Port names come from codec.Schema.Title (set via .WithTitle).
spec, err := pipeline.Render(reg.Spec())  // YAML pipeline document
fmt.Println(string(spec))

type myObserver struct{}

func (myObserver) RecordApply(name, version string, success bool, d time.Duration) {
    log.Printf("[forge] %s@%s ok=%v dur=%v", name, version, success, d)
}

// Lift scalar function over slice
mapToCelsius := forge.Map("mapToCelsius", "1.0.0", rawToCelsius,
    forge.WithRefinement(func(readings []RawReading) error {
        if len(readings) == 0 {
            return fmt.Errorf("batch must contain at least one reading")
        }
        return nil
    }),
)

// Errors attributed to element or key
_, err := mapToCelsius.Apply(batch)
var ce forge.CollectionElementError
if errors.As(err, &ce) {
    fmt.Printf("element %d failed in %q: %v\n", ce.Index, ce.Function, ce.Err)
}

pngCodec := codex.Bytes().
    Refine(validate.MaxBytes(5 * 1024 * 1024)).
    Refine(validate.PNG).
    WithTitle("rawImage")

// Validates input + output; PNG magic-byte check runs on both
resizeImage := forge.NewFunction("resizeImage", "1.0.0",
    pngCodec,
    pngCodec.WithTitle("resizedImage"),
    func(raw []byte) ([]byte, error) {
        return resizePNG(raw, 128, 128)
    },
    forge.FunctionMeta{Description: "Downscale PNG to 128×128 thumbnail."},
)

result, err := resizeImage.Apply(pngBytes)
// validate.PNG ran on pngBytes (input) and result (output)

measuredPNG := forge.MeasuredCodec(codex.Bytes().Refine(validate.PNG))