Go Framework

Module: github.com/redelay/go-framework

Full-featured Go framework providing a module system with auto-discovery, built-in auth/users/groups modules with self-contained HTTP routes, a transport-agnostic bootstrap system with pluggable runners (HTTP, Kafka, FlowDSL, gRPC, WebSocket), generic MongoDB and ClickHouse CRUD, ClickHouse batch writer, Redis-backed distributed primitives (lock, semaphore), JWT + password management, OpenAPI 3.0 spec generation with Swagger UI / Scalar / Redoc, MCP server, and config loaders.

The canonical IR types, FlowDSL parser, compilation pipeline, structural validation, dependency resolution, code generation, and SVG tooling have been extracted to the standalone go-flowdsl library (github.com/redelay/go-flowdsl). The framework integrates it without re-exporting — use go-flowdsl directly for IR/DSL work. AsyncAPI and OpenAPI format adapters remain in go-framework/core.

Bootstrap & Runner architecture

The framework separates application lifecycle from transport:

1. app.Bootstrap() — connects Mongo, discovers modules, starts them → returns *Instance
2. Transport runners (server.Server, future consumer.KafkaRunner, grpc.Server, etc.) implement app.Runner
3. app.Run(inst, runners...) — starts all runners in parallel, handles signals, shuts down cleanly

This means the same module system powers HTTP, Kafka consumers, FlowDSL workers, gRPC, and WebSocket — without any code changes in modules.

Quick start (HTTP only)

import (
    "github.com/redelay/go-framework/app"
    "github.com/redelay/go-framework/server"
)

inst, srv, err := server.Default()  // bootstrap + HTTP runner
app.Run(inst, srv)                  // signal handling + graceful shutdown

Multiple runners (HTTP + Kafka)

inst, err := app.Bootstrap()  // transport-agnostic

httpSrv := server.FromInstance(inst, server.LoadConfig())
kafkaRunner := consumer.NewKafkaRunner(inst, consumer.LoadConfig())  // future

app.Run(inst, httpSrv, kafkaRunner)  // both in parallel

No HTTP at all (consumer-only service)

inst, err := app.Bootstrap()
kafkaRunner := consumer.NewKafkaRunner(inst, consumer.LoadConfig())
app.Run(inst, kafkaRunner)

Runner interface

// Any transport adapter implements this:
type Runner interface {
    Run(ctx context.Context) error
}

Planned runners: server.Server (HTTP/REST), consumer.KafkaRunner, grpc.Server, ws.Server, flowdsl.Runner.

app package

Bootstrap

import "github.com/redelay/go-framework/app"

// Zero-config from env vars:
inst, err := app.Bootstrap()

// With EventBus and Redis:
inst, err := app.Bootstrap(
    app.WithEventBus(bus),   // *eventbus.EventBus from go-events
    app.WithRedis(rdb),      // *redis.Client
)

// Or with explicit config:
cfg := app.LoadBootstrapConfig()
inst, err := app.BootstrapWith(cfg)

// Instance fields:
inst.App         // *app.App
inst.MongoClient // *mongo.Client
inst.Logger      // *zap.Logger
inst.Config      // *app.BootstrapConfig

// Shutdown:
inst.Shutdown(ctx)

Run

// Start runners, wait for SIGINT/SIGTERM, shut down:
err := app.Run(inst, runner1, runner2, ...)

BootstrapConfig env vars

Server package (HTTP runner)

The server package is the HTTP/REST runner. It wires chi router, middleware, module routes, and MCP server.

Factory methods

Server methods

HTTP Config env vars

What the HTTP runner sets up

1. chi router with RealIP, RequestID, Recoverer, CORS
2. auth.OptionalAuth — parse JWT if present
3. {ROUTE_PREFIX}/health endpoint
4. All RoutesProvider module routes under ROUTE_PREFIX
5. OpenAPI spec at /openapi.json + API reference UI at /reference
6. MCP server at /mcp

httputil package

import "github.com/redelay/go-framework/httputil"

httputil.WriteJSON(w, http.StatusOK, data)        // JSON response
httputil.ReadJSON(r, &input)                       // decode request body
httputil.Error(w, http.StatusBadRequest, "detail") // {"detail": "..."}

App container (low-level)

For full manual control without Bootstrap:

import "github.com/redelay/go-framework/app"

application := app.New("myapp", logger)
application.DiscoverAndRegister(modules.ModuleDeps{DB: db, Logger: logger})
application.Start(ctx)
authMod := application.Registry().Get("auth").(*auth.Module)
application.Shutdown(ctx)

Module system

Core interfaces

// Every module must implement:
type Module interface {
    Manifest() *Manifest
    Startup(ctx context.Context) error
    Shutdown(ctx context.Context) error
}

// Optional interfaces:
type Configurable interface { Configure(registry *Registry) error }
type RoutesProvider interface { Routes(router Router) }
type MCPProvider interface { MCPTools() []MCPTool; MCPResources() []MCPResource }
type EventsProvider interface {
    Events() []*ir.Event                         // IR metadata — fed to modspec + AsyncAPI
    Consumers() []*ConsumerRegistration          // runtime consumer wiring
}

ModuleDeps

ModuleDeps is passed to every factory function. Fields that are nil when the corresponding option is not configured.

type ModuleDeps struct {
    DB       *mongo.Database  // always set if MONGO_URI is configured
    Logger   *zap.Logger      // always set
    EventBus EventBus         // nil when no transport is configured
    Redis    *redis.Client    // nil when REDIS_URL is not set
}

Wire optional deps at bootstrap:

import (
    "github.com/redelay/go-events/eventbus"
    "github.com/redelay/go-events/transport/kafka"
)

t, _ := kafka.New(kafka.Config{Brokers: []string{"kafka:9092"}})
bus := eventbus.New(t)
if err := bus.Start(ctx); err != nil { log.Fatal(err) }

inst, err := app.Bootstrap(
    app.WithEventBus(bus),
    // app.WithRedis(redisClient),
)

Important: Call bus.Start(ctx) before app.Bootstrap(). Start wires all registered consumer subscriptions to the transport — no messages are consumed until it is called.

EventBus interface

Modules receive a narrow EventBus interface in ModuleDeps — they can publish events but cannot register new consumers or inspect internal state:

// modules.EventBus — the interface injected into ModuleDeps.
type EventBus interface {
    Publish(ctx context.Context, msg *EventMessage) error
    Close() error
}

The concrete eventbus.EventBus in go-events adds Register, Start, Stop, UsePublish, and UseConsume. Modules only ever see the narrow interface; consumers returned from EventsProvider.Consumers() are registered on the full struct and the bus is started by the go-events/module lifecycle module (blank-import it; TRANSPORT picks the backend).

EventDefinition (go-events/typed)

Compile-time-safe event descriptor. Define once per event, publish from any handler. The canonical import is github.com/redelay/go-events/typed (implementation in go-framework/runtime/events).

import "github.com/redelay/go-events/typed"

type OrderCreatedPayload struct {
    OrderID string  `json:"order_id"`
    Total   float64 `json:"total"`
}

var OrderCreatedEvent = typed.EventDefinition[OrderCreatedPayload]{
    Name:       "order.created",
    EntityType: "order",
    Action:     "created",
    Topic:      "order.created",
}

// Publish from a handler (deps.EventBus injected at bootstrap):
msg, _ := OrderCreatedEvent.NewMessage(
    order.ID,
    typed.Actor{Type: typed.ActorTypeUser, ID: userID},
    OrderCreatedPayload{OrderID: order.ID, Total: order.Total},
)
deps.EventBus.Publish(ctx, msg)

Actor types: typed.ActorTypeUser, ActorTypeAdmin, ActorTypeSystem, ActorTypeService; typed.SystemActor is Actor{Type: ActorTypeSystem, ID: "system"}. MustNewMessage panics instead of returning an error.

Well-known definitions:

EventsProvider

Implement EventsProvider when a module publishes or consumes events. It serves two purposes:

1. IR metadata — Events() feeds modspec and AsyncAPI spec generation
2. Runtime wiring — Consumers() is called by the app to register handlers before bus.Start()

func (m *OrderModule) Events() []*ir.Event {
    return []*ir.Event{
        {Name: "order.created", EntityType: "order", Action: "created"},
    }
}

func (m *OrderModule) Consumers() []*modules.ConsumerRegistration {
    return []*modules.ConsumerRegistration{
        {
            EventName: "payment.completed",
            Topic:     "payment.completed",
            GroupID:   "orders",
            Handler:   m.handlePaymentCompleted,
        },
    }
}

The ConsumerRegistration fields:

Manifest

&modules.Manifest{
    Name:        "mymodule",
    Description: "My custom module",
    Version:     "1.0.0",
    DependsOn:   []string{"auth"},     // started after auth
    Provides:    []string{"mymodule"}, // what this module provides
}

Auto-discovery

Modules register factories in init():

func init() {
    modules.RegisterFactory("mymodule", func(deps modules.ModuleDeps) (modules.Module, error) {
        return NewModule(deps.DB, deps.Logger), nil
    })
}

Blank-import the package to trigger registration:

import _ "github.com/redelay/go-framework/modules/auth"

Registry — dual-key indexing by Name and ID

Registry.Register indexes each module by both manifest.Name and manifest.ID when they differ. This lets admin sibling modules reach the core module by its canonical ID regardless of how the original author capitalised the Name field:

// core module
&modules.Manifest{ ID: "users", Name: "Users" }

// admin sibling — Configure hook
func (m *Module) Configure(r *modules.Registry) error {
    peer, _ := r.Get("users")          // lookup by manifest.ID — always works
    m.core = peer.(*usersmod.Module)
    return nil
}

manifest.ID defaults to manifest.Name when unset, so legacy modules that only set Name keep working with no changes. When both are set and differ, both keys resolve to the same Module pointer; the ID alias loses to an existing entry with the same key (Name always wins on collision).

Public / admin split convention

Every module that exposes mutating HTTP endpoints splits into two packages:

The factory name convention is <module>-admin. The admin package does not re-register the core module's factory — it depends on the core module being imported too (depends_on: <module> in module.yaml). Admin handlers reach core state via accessor methods on the core Module struct (Logger, Service, Store, etc.), or by asking the Registry for the peer module.

Admin routes sit under the AdminPrefix() (/admin by default, override via ADMIN_ROUTE_PREFIX) and are wrapped in:

r.Group(modules.AdminPrefix()+"/<name>", func(g modules.Router) {
    g.Use(auth.RequireActiveUser, auth.RequirePermission("admin:access"))
    // ... admin writes ...
})

This convention is why the backend ships two binaries: cmd/api pulls in core modules only (never the admin/ siblings), so its route table simply cannot respond to admin-only mutations. cmd/admin-api pulls both and lives behind the operator's VPN / Tailscale.

Module inventory — shared bundles per profile

Every backend binary blank-imports exactly one or two "bundle" packages that enumerate which modules land in that profile. The bundles live under backend/imports/:

Adding a new module is a one-line change:

1. Shared by api + admin-api → add blank-import to backend/imports/common/common.go.
2. Admin-only HTTP surface (has foo/admin/ with id: foo-admin) → add to backend/cmd/admin-api/admin_imports.go.
3. CLI-only command provider → add to backend/imports/cli/cli.go.

No other file needs editing. Each main.go is just five lines of wiring plus the bundle imports.

Leak enforcement — three layers, same intent. No route or middleware from an admin package can reach cmd/api's surface. Each layer catches a class of mistake the others miss, so they complement rather than duplicate.

Adding an admin blank-import to cmd/api, to backend/imports/common, or to any transitive dependency of either, trips one of these tests with a human-readable message listing the leaking paths / routes.

Why a shared package instead of duplicated lists per binary? Two or three binaries keeping parallel import lists in sync is a known failure mode — someone adds a module to cmd/api/main.go and forgets cmd/admin-api/main.go, then prod admin-api silently misses a feature. The bundle pattern makes the mistake structurally impossible: adding a module means editing the one file that both binaries already import.

When the shared inventory grows past ~60 imports or the admin inventory past ~20, a go generate tool that scans */module.yaml and regenerates these bundles becomes worthwhile. At current scale the manual lists are simpler to audit and diff.

See the audit in any core module (e.g. users, assistant, flowexec) for the concrete public/admin split.

Testing & CI guardrails {#testing-ci-guardrails}

The monorepo has many Go modules (go-framework, go-flowdsl, go-events, go-modules, go-module-email, go-ai, backend, …). infra/Makefile provides single-command entry points that iterate the full set so CI and pre-push checks never miss a module:

Recent hardening worth knowing about:

• openapi.joinRoutePrefix (go-framework/core/openapi/spec.go) joins two route-path segments without ever producing //. Locked in by TestJoinRoutePrefix and TestCollectRoutes_NoDoubleSlashInRealPattern — these catch the regression class the make check-openapi-paths target guards against at the spec level.
• Router adapter Group("/", fn) (go-framework/server/adapter.go) now uses chi.Router.Group (middleware-only inline subrouter, no mount) when the prefix is empty. Multiple modules can apply admin gating via r.Group("/", g.Use(auth.RequireActiveUser, auth.RequirePermission("admin:access"))) on the same parent router without colliding on chi's mount-point check. Non-empty prefixes still use chi.Router.Route.

Built-in modules

auth

Factory name: "auth" · DependsOn: ["users"] · Provides: ["auth", "jwt", "tokens"] Implements: RoutesProvider (login, refresh, revoke under /auth), MiddlewareProvider

Routes (auto-registered):

The login endpoint supports both application/json ({\"email\", \"password\"}) and application/x-www-form-urlencoded (username=&password=) for OAuth2 password flow compatibility with Swagger UI.

Configurable mount prefix. The auth routes mount at AUTH_ROUTE_PREFIX (default /auth, relative to ROUTE_PREFIX). A project porting a legacy surface points it elsewhere — e.g. AUTH_ROUTE_PREFIX=/login — and every auth route moves with it, no per-handler change.

Extended login surface (passwordless + TOTP). Beyond password login, the module registers a fuller authentication surface. All paths are relative to AUTH_ROUTE_PREFIX.

TOTP (totp.go) is RFC 6238, stdlib-only: it accepts a ±1-period clock skew and compares codes in constant time. A login for a TOTP-enrolled user returns a short-lived claim token instead of a full session; the client completes it at POST /totp, and the issued Claims.TOTPValid records that the second factor was satisfied.

Claim tokens (claim.go) are purpose-scoped, short-lived JWTs — a token minted for totp cannot be redeemed as a magic, verify, or reset token. Service.ParseClaim(tokenStr, purpose) enforces the purpose, so a leaked magic-link token cannot be repurposed to reset a password.

API schemas:

Service methods:

Claims struct:

type Claims struct {
    UserID      string   `json:"sub"`
    Email       string   `json:"email"`
    IsSuperuser bool     `json:"is_superuser"`
    TokenType   TokenType
    TOTPValid   bool
    Permissions []string `json:"permissions,omitempty"`
    Extra       map[string]any
    jwt.RegisteredClaims
}

Password functions:

Middleware:

Context helpers:

ctx = auth.ContextWithClaims(ctx, claims)
claims, ok := auth.ClaimsFromContext(ctx)

Token source precedence — AuthMiddleware (and OptionalAuth) looks for the access token in two places, in order:

1. Authorization: Bearer <jwt> header (canonical).
2. ?access_token=<jwt> query-string fallback (RFC 6750 §2.3 / OAuth2 SSE idiom). Needed because the browser EventSource API cannot set custom headers — SSE endpoints like /runs/{id}/events or /flows/{id}/live would otherwise be unreachable from a browser without proxying the token through a cookie.

If both are present, the header wins. Deployments that terminate TLS at the edge should ensure access logs redact the access_token query parameter. Regression tests: TestAuthMiddleware_QueryParamFallback, TestAuthMiddleware_HeaderWinsOverQueryParam.

Cross-module interfaces:

// Implement in your users module:
type LookupProvider interface {
    AsAuthLookup() UserLookup
}

// Implement in your groups module (optional):
type PermissionsProvider interface {
    AsPermissionsResolver() PermissionsResolver
}

users

Factory name: "users" · Provides: ["users"] Implements: RoutesProvider (self-service under /users), auth.LookupProvider

Public routes (registered by core users module — reach any binary that blank-imports it):

Admin routes (registered by the users/admin sibling module — blank-import only in admin-api):

Every admin route is gated on admin:access + auth:active-user. The public api binary has no write surface for other users; the admin api binary exposes both the public self-service routes AND the admin administration routes.

Service methods:

API schemas:

Implements auth.LookupProvider automatically.

groups

Factory name: "groups" · Provides: ["groups", "permissions"]

Service methods:

Implements auth.PermissionsProvider automatically.

health

Factory name: "health" · Implements RoutesProvider, MountProvider, DynamicRoutesProvider.

Serves the same health check at two places:

The root path is HEALTH_ROOT_PATH (default /health); set it empty to disable the root mount. It is advertised in /openapi.json at its absolute path via DynamicRoutes + AbsolutePath. The response aggregates every check registered in the shared HealthRegistry (Mongo, Redis, and any a module adds), returning 200 when all pass and 503 when any fails.

FlowDSL node subpackages

Each built-in module (auth, users, groups) has a flowdsl/ subdirectory providing FlowDSL nodes as a separate companion module. Enable with a blank import:

import (
    _ "github.com/redelay/go-framework/modules/auth"
    _ "github.com/redelay/go-framework/modules/auth/flowdsl"   // 7 FlowDSL nodes
)

For the full node catalog and subpackage structure, see the FlowDSL node subpackage reference.

Generic CRUD

MongoDB CRUD

import "github.com/redelay/go-framework/crud"

type MyModel struct {
    crud.BaseModel `bson:",inline"`
    Name string    `bson:"name" json:"name"`
}

func (m *MyModel) GetID() primitive.ObjectID { return m.ID }

c := crud.NewMongoCRUD[*MyModel](collection)

item, _ := c.Create(ctx, &MyModel{Name: "test"})
item, _ = c.FindByID(ctx, id)
item, _ = c.FindByField(ctx, "name", "test")
result, _ := c.List(ctx, crud.ListParams{
    Filter: bson.M{"name": "test"},
    Skip: 0, Limit: 20,
    Sort: bson.D{{Key: "created_at", Value: -1}},
})
// result.Items, result.Total, result.Skip, result.Limit

Projections — ListParams.Projection

Projection takes the driver's usual bson.M{"field": 1} / {"field": 0} form and is nil by default, which reads whole documents.

Reach for it whenever the documents are much larger than the response. The motivating case is localized content: a catalogue whose name, description and step lists hold every locale is roughly 39 KB per document, so a 50-item page pulls ~1.9 MB off the server to render a grid of names and thumbnails.

Prefer an exclusion projection ({"field": 0}):

• An inclusion list has to enumerate every other field, so a field added to the model later comes back empty in list views with nothing to say why.
• Exclusion cannot damage a legacy document shape. If instructions is still a plain array on old rows rather than a per-locale map, naming it in an exclusion projection still just removes it, whereas including instructions.en silently yields [].

Move the DTO with the projection. A projected-away field decodes as its zero value, which is indistinguishable from "this record genuinely has none" — so a list DTO should nil those fields and mark them omitempty, leaving the key absent. undefined then means "not in this view" and [] means empty. Test the pair together; each half is silent on its own.

Read preference — per query, not per deployment

The client default is primary and should stay there. A global secondary preference is the tempting setting and the wrong one: replication is asynchronous, so it breaks read-your-writes on every endpoint at once — a user writes, the next poll hits a lagging member, and the write appears to vanish.

Whether a read tolerates staleness is a property of the read, so it is opted into at the call site:

import rtmongo "github.com/redelay/go-framework/runtime/mongo"

// Curated content, an analytics rollup — anything the caller did not just write.
ctx = rtmongo.Nearest(ctx)          // closest member, including the primary
ctx = rtmongo.Secondary(ctx)        // keep heavy reporting off the primary
items, err := store.List(ctx, params)

crud.MongoCRUD honours it on reads only — a read preference on a write is meaningless, and routing writes through the same seam would imply otherwise. A context with no preference uses the client default, so existing code is unaffected.

Secondary maps to SecondaryPreferred deliberately: on a single-node deployment (every dev machine) a hard Secondary has nowhere to go and fails the query, where preferring one degrades to the primary.

Lifecycle events — WithEvents

Any CRUD can publish <entity>.created / .updated / .deleted after successful writes, so other modules react to data changes without importing the owner:

c := crud.NewMongoCRUD[*WorkoutProgram](col).WithEvents(deps.EventBus, "program")

That is the whole opt-in. Every Create, Update and Delete then publishes an EventMessage with EntityType: "program", the entity id, and Headers["topic"] = "program.updated" — the same routing convention the built-in domain modules use, so flows and EventsProvider consumers subscribe by topic name.

• No transport configured? WithEvents(nil, …) is a no-op — wire deps.EventBus unconditionally.
• Publish failures never fail the write. The write already succeeded; the event is best-effort.
• The payload is {entity_type, entity_id}. Consumers that need the full document re-read it by id — by the time they run it may have changed again, so the id is the only durable handle.

Typical consumer: a search module re-embedding exactly the document that changed (see Search — keeping an index fresh).

Per-user singleton — UserSingleton[T]

"Preferences", "profile", "settings for me" and their relatives are the same resource shape everywhere: at most one document per user, every field defaulted, and a reset that returns to those defaults. Written as ordinary CRUD each time, they pick up the same three bugs — a 404 for a user who has simply never saved, a partial update that clobbers unset fields, and a reset that leaves a stale row behind. crud.UserSingleton[T] is that shape, done once.

type Prefs struct {
    crud.BaseModel `bson:",inline"`
    UserID      string `bson:"user_id"`
    RestSeconds int    `bson:"rest_seconds"`
    WeightUnit  string `bson:"weight_unit"`
}

func (p *Prefs) SetUserID(id string) { p.UserID = id }   // satisfies crud.UserOwned

func defaults(userID string) *Prefs {
    return &Prefs{UserID: userID, RestSeconds: 90, WeightUnit: "kg"}
}

store := crud.NewUserSingleton[*Prefs](db.Collection("prefs"), defaults)
_ = store.EnsureIndexes(ctx, "prefs_user_unique")   // one row per user

The defining behaviour is that Get never 404s: a user who has never opened the settings screen gets defaults, not an error, and the caller need not know the difference.

p, stored, _ := store.Get(ctx, userID)          // stored=false → these are defaults
p, _         = store.Upsert(ctx, userID, bson.M{"rest_seconds": 45})  // partial $set
p, _         = store.Reset(ctx, userID)         // deletes the row, returns defaults
ok, _       := store.Delete(ctx, userID)        // reports whether one existed

Upsert seeds the document from defaults before applying the patch, so a one-field save does not write zero values across every other field. Reset deletes rather than overwriting with defaults — a row full of default values is indistinguishable from a user who deliberately chose them, so a reset should leave no trace.

ClickHouse CRUD

Generic CRUD for ClickHouse analytics tables. Uses ch struct tags for column mapping (falls back to lowercase field name). Requires the ClickHouseModel interface:

import (
    "github.com/redelay/go-framework/crud"
    "github.com/redelay/go-framework/runtime/clickhouse"
)

type PageView struct {
    Timestamp time.Time `ch:"timestamp"`
    Domain    string    `ch:"domain"`
    Path      string    `ch:"path"`
    UserAgent string    `ch:"user_agent"`
    Count     uint64    `ch:"count"`
}

func (p *PageView) TableName() string { return "page_views" }

conn, _ := clickhouse.NewConn(cfg, "analytics", nil)
c := crud.NewClickHouseCRUD(conn, func() *PageView { return &PageView{} })

// Single insert
c.Create(ctx, &PageView{Domain: "example.com", Path: "/", Count: 1})

// Bulk insert (preferred for ClickHouse)
c.BulkCreate(ctx, views)  // uses PrepareBatch + AppendStruct

// Read
view, _ := c.Get(ctx, "domain", "example.com")
views, _ := c.List(ctx, crud.CHListParams{
    Filters:  map[string]any{"domain": "example.com"},
    OrderBy:  "timestamp DESC",
    Limit:    100,
})
count, _ := c.Count(ctx, map[string]any{"domain": "example.com"})

// Raw query
rows, _ := c.Query(ctx, "SELECT * FROM page_views WHERE domain = ?", "example.com")

Available methods: Create, BulkCreate, Get, GetByField, List, Count, Query.

ClickHouse batch writer

For high-throughput scenarios (e.g. Kafka consumers writing analytics), the batch writer queues items in memory and flushes in bulk — either when batchSize rows accumulate or flushInterval elapses, whichever comes first. Thread-safe.

// Backed by a ClickHouseCRUD
writer := crud.BatchWriterFor(c,
    crud.WithBatchSize(500),
    crud.WithFlushInterval(5 * time.Second),
    crud.WithBatchName("pageviews"),
)

// From any goroutine — non-blocking
writer.Enqueue(&PageView{Domain: "example.com", Path: "/"})

// On shutdown — drains remaining items
writer.Stop()

You can also supply a custom flush function for non-CRUD use cases:

writer := crud.NewClickHouseBatchWriter(func(ctx context.Context, items []*PageView) (int, error) {
    // custom bulk insert logic
    return len(items), nil
}, crud.WithBatchSize(1000))

OpenAPI 3.0 spec generation

The openapi package generates a full OpenAPI 3.0.3 specification from module routes, including typed parameters, request/response schemas via $ref, validation constraints, and configurable security schemes.

Automatic spec + UI (via openapi module)

The openapi built-in module serves the spec and an interactive API reference UI. No code required — blank-import it:

import _ "github.com/redelay/go-framework/modules/openapi"

This adds:

UI provider selection

OpenAPI environment variables

Endpoint metadata (functional options)

Routes declare OpenAPI metadata via functional options on Router.Handle():

func (m *Module) Routes(router modules.Router) {
    router.Group("/users", func(r modules.Router) {
        r.Handle("POST", "/", http.HandlerFunc(m.handleCreate),
            modules.Summary("Create user"),
            modules.Description("Create a new user. Requires superuser access."),
            modules.Body(UserCreateInput{}),
            modules.Response(201, "User created", UserResponse{}),
            modules.Response(400, "Invalid request body", modules.RedelayErrorResponse{}),
            modules.Response(409, "User already exists", modules.RedelayErrorResponse{}),
            modules.Security("BearerAuth"),
        )
    })
}

Available options:

Schema generation

Go structs are reflected into JSON Schema automatically:

• Struct field names from json:"..." tags
• validate:"required" → JSON Schema required array
• validate:"email" → format: "email"
• validate:"min=8" → minLength: 8 (strings) or minimum: 8 (numbers)
• *T pointer fields → nullable: true
• time.Time → format: "date-time"
• primitive.ObjectID → format: "objectid"
• []T → type: "array" with typed items
• Embedded structs merge properties

Named structs are registered in components/schemas and referenced via $ref.

Security schemes

openapi.GenerateSpec(title, version, registry,
    openapi.WithBearerAuth(),                                    // HTTP Bearer/JWT
    openapi.WithOAuth2PasswordFlow("/api/v1/auth/login", nil),   // OAuth2 password flow
)

The OAuth2 password flow enables Swagger UI's "Authorize" popup with username/password fields.

Schema naming convention

All API schemas use module-prefixed names to avoid collisions across modules:

MCP server

Automatically mounted at /mcp by the server package. Implements the MCP Streamable HTTP transport (protocol version 2025-03-26) — compatible with VS Code, Claude Desktop, Cursor, and other MCP clients.

For manual use:

import "github.com/redelay/go-framework/server/mcp"

server := mcp.NewServer()
server.DiscoverFrom(registry) // auto-discovers MCPProvider modules + routes
router.Mount("/mcp", server.Handler())

Protocol support

The MCP server handles the full MCP lifecycle:

IDE setup

Add to your .vscode/mcp.json:

{
  "servers": {
    "my-app": {
      "type": "http",
      "url": "http://localhost:8000/mcp"
    }
  }
}

Once connected, AI assistants can discover and invoke all registered tools and resources.

Built-in introspection tools

DiscoverFrom installs a family of introspection tools for AI assistants (Claude, Cursor, Copilot, etc.) to reason about the running application. All tools read live from the module registry — no caching.

Developer tools

Additional tools for building and validating modules and flows:

Module-contributed tools

Any module can contribute additional MCP tools by implementing the MCPProvider interface. These are discovered automatically by DiscoverFrom.

For example, the flowexec addon module contributes:

See MCPProvider for how to add tools from your own modules.

Built-in resources

MCPProvider

Modules expose tools and resources to the MCP server by implementing MCPProvider:

type MCPProvider interface {
    MCPTools() []MCPTool
    MCPResources() []MCPResource
}

Example — exposing a search tool from a custom module:

func (m *Module) MCPTools() []modules.MCPTool {
    return []modules.MCPTool{
        {
            Name:        "search_products",
            Description: "Search the product catalog by name or category",
            InputSchema: map[string]any{
                "type":     "object",
                "required": []string{"query"},
                "properties": map[string]any{
                    "query":    map[string]any{"type": "string", "description": "Search term"},
                    "category": map[string]any{"type": "string", "description": "Filter by category"},
                },
            },
            Handler: func(ctx context.Context, args map[string]any) (any, error) {
                query, _ := args["query"].(string)
                return m.service.Search(ctx, query)
            },
        },
    }
}

func (m *Module) MCPResources() []modules.MCPResource { return nil }

Once the module is registered, its tools appear automatically in tools/list responses.

Config

Helper functions

Pre-built config loaders

KafkaConfig

cfg.BrokerAddrs() returns []string from the comma-separated Brokers field — ready for sarama.

NATSConfig

cfg.URLs() splits the cluster list into []string.

RedisStreamsConfig

cfg.Addr() extracts host:port from the URL for go-redis.

Runtime

Redis

client, err := redis.NewClient(cfg)
key := redis.Key(cfg, "inflight", domain)  // → "prefix:inflight:example.com"

ClickHouse

conn, err := clickhouse.NewConn(cfg, "service-name", nil)

Cross-container coordination (runtime/coordination)

Transport-agnostic coordination primitives for multi-container deployments — pub/sub invalidation, append-only streams for fan-out, and watchable KV storage. Blank-import go-modules/coordination/module and set COORD=nats|redis|noop to auto-wire into ModuleDeps.Coordinator. Zero external deps in the interface package; backends live in go-modules/coordination/{nats,redis}.

if c := m.deps.Coordinator; c != nil {
    _ = c.Publish(ctx, "flow.published", []byte(flowID))   // invalidation
    _, _ = c.AppendStream(ctx, "live:"+flowID, frameJSON)  // fan-out
    _ = c.KVSet(ctx, "routes", flowID, routeJSON)          // shared config
    w, _ := c.KVWatch(ctx, "routes", "*")                  // watch for updates
}

See the full Coordination reference for the interface, backends, and conformance test suite.

Distributed primitives

Redis-backed coordination primitives in runtime/distributed. Both degrade gracefully when Redis is nil — the lock always reports leader, the semaphore always grants.

Distributed lock (leader election)

Binary mutex backed by SET NX EX with Lua CAS release. Runs a background heartbeat to renew the lock automatically.

import "github.com/redelay/go-framework/runtime/distributed"

lock := distributed.NewDistributedLock(redisClient, "worker-leader",
    distributed.WithLockTTL(120 * time.Second),
    distributed.WithHeartbeatInterval(30 * time.Second),
    distributed.WithLockPrefix("myapp:dlock:"),
)

lock.Start(ctx)          // tries to acquire; runs heartbeat in background
defer lock.Stop(ctx)     // releases lock

if lock.IsLeader() {
    // only one instance runs this
}

info, _ := lock.GetCurrentLeader(ctx) // {"token": "...", "is_self": true/false}

Distributed semaphore

Counting semaphore backed by a Redis sorted set with expiry scores. Supports configurable max workers and heartbeat renewal.

sem := distributed.NewDistributedSemaphore(redisClient, "import-workers", 3,
    distributed.WithSemTTL(300 * time.Second),
    distributed.WithSemHeartbeat(60 * time.Second),
    distributed.WithAcquireTimeout(30 * time.Second),
    distributed.WithSemPrefix("myapp:dsem:"),
)

// Manual acquire/release
token, _ := sem.Acquire(ctx)
defer sem.Release(ctx, token)

// Or use Hold — acquires, runs fn with background heartbeat, releases
sem.Hold(ctx, func(ctx context.Context) error {
    // do work within semaphore slot
    return nil
})

count, _ := sem.CurrentCount(ctx) // active holders

Core IR (Intermediate Representation)

The ir package defines the canonical data model that all formats import into and export from. Every FlowDSL, OpenAPI, AsyncAPI, or business model document is first converted to this IR before any processing.

Document

import "github.com/redelay/go-flowdsl/ir"

doc := ir.NewDocument("My App", "1.0.0")
// doc.ID       — auto-generated UUID
// doc.Modules  — business/domain modules
// doc.Entities — domain aggregates with typed fields
// doc.Events   — business events (entity_type + action)
// doc.Commands — state-changing intents
// doc.Queries  — read operations
// doc.Actions  — top-level operations (command/query/mutation/subscription)
// doc.Workflows — multi-step processes with nodes + edges
// doc.Channels — message/event channels with pub/sub
// doc.Schemas  — reusable data structure definitions
// doc.Packets  — data transfer payloads
// doc.Schedules — time-based triggers (cron/interval)
// doc.Migrations — database migration definitions
// doc.Metadata — arbitrary key-value metadata
// doc.Provenance — origin and mapping history

Key types

FlowDSL

The flowdsl package parses, validates, normalizes, and exports FlowDSL — the primary DSL for defining Redelay applications.

Parse & Validate

import "github.com/redelay/go-flowdsl/flowdsl"

parsed, err := flowdsl.Parse(reader)       // YAML → *FlowDSL
diag := flowdsl.Validate(parsed)           // structural checks → *diagnostics.Collector
doc := flowdsl.Normalize(parsed)           // *FlowDSL → *ir.Document
err = flowdsl.Export(doc, writer)           // *ir.Document → FlowDSL YAML

FlowDSL YAML format

version: "1.0"
title: My Application
modules:
  billing:
    name: billing
    description: Billing domain
    depends_on: [module.users]
    provides: [invoices]
entities:
  invoice:
    name: invoice
    fields:
      amount: { type: float, required: true }
      currency: { type: string, default: USD }
events:
  invoice_created:
    name: invoice_created
    entity_type: invoice
    action: created
    topic: billing.invoices
actions:
  create_invoice:
    name: create_invoice
    kind: command
    method: POST
    path: /invoices
workflows:
  payment_flow:
    name: payment_flow
    triggers: [invoice_created]
    steps:
      - id: validate
        kind: action
        action: validate_payment
        next: charge
      - id: charge
        kind: action
        action: charge_card
channels:
  notifications:
    name: notifications
    protocol: kafka
schedules:
  daily_report:
    name: daily_report
    cron: "0 9 * * *"
    action: generate_report

Validation rules

Compile pipeline

The core/compile package orchestrates the full import → validate → resolve → enrich → export pipeline, including AsyncAPI and OpenAPI format adapters.

Formats

Functions

import "github.com/redelay/go-framework/core/compile"

// Import a source format into IR
doc, err := compile.Import(compile.FormatFlowDSL, reader)

// Run validation + resolution + enrichment
result := compile.Compile(doc)
// result.Document    — enriched IR document
// result.Diagnostics — *diagnostics.Collector with errors/warnings
// result.Modules     — dependency-ordered modules
// result.HasErrors() — true if compilation produced errors

// One-step: import + compile
result, err := compile.ImportAndCompile(compile.FormatFlowDSL, reader)

// Export IR back to a format
err = compile.Export(doc, compile.FormatFlowDSL, writer)

Compilation stages

1. Validate — structural correctness (IDs, refs, module naming)
2. Resolve — topological sort of modules by dependencies
3. Enrich — promote module-level entities/events to document level, warn on unresolved provides

Diagnostics

The diagnostics package provides structured error/warning collection used throughout the compilation pipeline.

import "github.com/redelay/go-flowdsl/diagnostics"

c := diagnostics.NewCollector()
c.Error("source", "CODE", "message")
c.Warning("source", "CODE", "message")
c.Info("source", "CODE", "message")

c.HasErrors()  // bool
c.Errors()     // []*Diagnostic (error-level only)
c.Warnings()   // []*Diagnostic (warning-level only)
c.All()        // []*Diagnostic (all levels)

Each Diagnostic has: Severity (error/warning/info/hint), Code, Message, Source, and optional File, Line, Column, Pointer, Related.

Validate

The validate package checks an IR Document for structural correctness and consistency.

import "github.com/redelay/go-flowdsl/validate"

diag := validate.Validate(doc) // → *diagnostics.Collector

Validation rules

Resolver

The resolver package performs topological sorting of modules based on dependencies using Kahn's algorithm with cycle detection.

import "github.com/redelay/go-flowdsl/resolver"

sorted, diag := resolver.Resolve(doc.Modules)
// sorted — modules in dependency order (dependencies first)
// diag — errors for duplicates, unknown deps, cycles

Metadata

The metadata package extracts and stamps document metadata.

import "github.com/redelay/go-flowdsl/metadata"

info := metadata.Extract(doc)
// info.Title, info.Version, info.Compiler
// info.ModuleCount, info.EntityCount, info.EventCount
// info.ActionCount, info.WorkflowCount, info.ChannelCount, info.ScheduleCount

metadata.Stamp(doc) // adds compiler version + compiled_at to doc.Metadata

names := metadata.ModuleNames(doc)  // []string
names = metadata.ActionNames(doc)   // []string
names = metadata.EventNames(doc)    // []string

AsyncAPI import/export

The core/asyncapi package imports AsyncAPI 2.x YAML documents and exports IR Documents to AsyncAPI 2.6 YAML.

Import

import "github.com/redelay/go-framework/core/asyncapi"

doc, err := asyncapi.ImportYAML(reader) // AsyncAPI YAML → *ir.Document

Converts channels to ir.Channel with publish/subscribe events, Kafka group ID bindings.

Export

err := asyncapi.Export(doc, writer) // *ir.Document → AsyncAPI 2.6 YAML

Export behavior:

• IR channels with publish/subscribe events → AsyncAPI channels
• If no channels exist but events are present, creates synthetic channels using entity_type.action as topic name
• Kafka bindings for group IDs

Business model import/export

The businessmodel package parses business model YAML and converts to/from IR.

Import

import "github.com/redelay/go-flowdsl/businessmodel"

spec, err := businessmodel.ParseYAML(reader) // YAML → *Spec
doc := businessmodel.Normalize(spec)          // *Spec → *ir.Document

Export

err := businessmodel.Export(doc, writer) // *ir.Document → business model YAML

Export behavior:

• If modules exist, each module becomes a domain with entities, events, and processes (workflows)
• If no modules exist, creates a single domain from top-level items

Business model YAML format

title: E-Commerce Platform
version: "1.0"
domains:
  - name: billing
    description: Billing domain
    entities:
      - name: invoice
        fields: [amount, currency, status]
    events:
      - name: invoice_created
        entity_type: invoice
        action: created
    processes:
      - name: payment_flow
        triggers: [invoice_created]
        steps: [validate, charge, confirm]

OpenAPI tool schema export

The core/openapi package also provides functions for exporting IR Documents as function-calling tool definitions (JSON Schema format compatible with LLM tool-use APIs).

Export functions

import "github.com/redelay/go-framework/core/openapi"

tools := openapi.ExportActions(doc)   // IR actions → tool definitions
tools = openapi.ExportEntities(doc)   // IR entities → get_<name> lookup tools
tools = openapi.ExportCommands(doc)   // IR commands → tool definitions
tools = openapi.ExportAll(doc)        // all of the above combined
err := openapi.WriteTools(writer, tools) // serialize as JSON

Output format

[
  {
    "type": "function",
    "function": {
      "name": "create_invoice",
      "description": "Create a new invoice",
      "parameters": {
        "type": "object",
        "properties": {
          "amount": { "type": "number", "description": "Invoice amount" },
          "currency": { "type": "string" }
        },
        "required": ["amount"]
      }
    }
  }
]

Type mapping (IR → JSON Schema)

Supports nested properties, items for arrays, enum, format, and required fields.

CLI

There are two binaries named redelayctl in the Redelay source tree, with non-overlapping responsibilities. Which one you invoke depends on whether you're shaping a spec or operating a live app:

Both names are intentional: one works on source files, the other works against a running process. They never share an import graph.

Project redelayctl — contributing commands from a module

Any module that satisfies the CLIProvider interface gets its commands auto-discovered at bootstrap and exposed through ./cmd/redelayctl:

type CLIProvider interface {
    CLICommands() []CLICommand
}

type CLICommand struct {
    Name, Description string
    Args    []CLIArg
    Handler func(ctx context.Context, args map[string]string) error
}
type CLIArg struct {
    Name, Description, Default string
    Required, Flag             bool
}

Implement CLICommands() on the module struct, blank-import the module in backend/imports/cli/cli.go, and the command shows up in redelayctl help. The framework also auto-exposes every command as an MCP tool (prefixed cli_) so AI assistants can invoke the same operations.

Invoke via the Makefile helper (loads .env, resolves module bundle):

make redelayctl ARGS="help"
make redelayctl ARGS="create-user [email protected] --password=secret --superuser"
make redelayctl ARGS="assistant-ingest-docs --docs-dir ../spec/docs --url-base https://redelay.com/docs"

make cli is kept as an alias for scripts pinned to the old name.

Currently registered commands (expand as modules declare more):

Dockerfiles ship both /redelayctl and a /cli alias; the alias will be removed in a future release once scripts have migrated.

Framework redelayctl — spec tooling

redelayctl (the framework binary) is the command-line tool for working with Redelay specifications.

Commands

# Validate a FlowDSL file (runs FlowDSL + IR validation)
redelayctl validate app.flow.yaml

# Import a source format and output IR as JSON
redelayctl import openapi api.yaml
redelayctl import asyncapi events.yaml
redelayctl import businessmodel model.yaml

# Export IR to a target format
redelayctl export json app.ir.yaml      # IR → JSON
redelayctl export yaml app.ir.json      # IR → YAML
redelayctl export flowdsl app.ir.json   # IR → FlowDSL YAML
redelayctl export openapi app.ir.json   # IR → OpenAPI tool schema JSON

# Convert between formats directly
redelayctl convert flowdsl json app.flow.yaml
redelayctl convert openapi flowdsl api.yaml
redelayctl convert asyncapi openapi events.yaml

# Parse FlowDSL and output canonical IR
redelayctl ir app.flow.yaml

# Print version
redelayctl version

Supported format conversions