Documentation
¶
Overview ¶
Package flowapi is the operator-facing Flow surface: realm-scoped pipeline CRUD (store) and named durable flow lifecycle (store + flow.Manager). Paths are Astrate-native (/flow/v1/...) until a real upstream client forces wire-compatible routes (milestone v2.0 gap 3).
Index ¶
- Variables
- type API
- type CreateFlowRequest
- type FlowView
- type PipelineView
- type Service
- func (s *Service) CreateAndStartFlow(ctx context.Context, realm string, req CreateFlowRequest) (*FlowView, error)
- func (s *Service) CreatePipeline(ctx context.Context, realm, name string, definition []byte) (*PipelineView, error)
- func (s *Service) CreateUserBlock(ctx context.Context, realm, name, blockType string, ...) (*UserBlockView, error)
- func (s *Service) DeleteFlow(ctx context.Context, realm, name string) error
- func (s *Service) DeletePipeline(ctx context.Context, realm, name string) error
- func (s *Service) DeleteUserBlock(ctx context.Context, realm, name string) error
- func (s *Service) GetBlock(_, blockType string) (*blocks.Info, error)
- func (s *Service) GetFlow(ctx context.Context, realm, name string) (*FlowView, error)
- func (s *Service) GetPipeline(ctx context.Context, realm, name string) (*PipelineView, error)
- func (s *Service) GetUserBlock(ctx context.Context, realm, name string) (*UserBlockView, error)
- func (s *Service) HasBuiltInBlock(blockType string) bool
- func (s *Service) ListBlocks(_ string) []blocks.Info
- func (s *Service) ListFlows(ctx context.Context, realm string) ([]FlowView, error)
- func (s *Service) ListPipelines(ctx context.Context, realm string) ([]string, error)
- func (s *Service) ListUserBlocks(ctx context.Context, realm string) ([]UserBlockView, error)
- func (s *Service) Manager() *flow.Manager
- func (s *Service) MarkRunningFlowsStopped(ctx context.Context)
- func (s *Service) RehydrateAutoRestart(ctx context.Context) error
- func (s *Service) ReloadFlow(ctx context.Context, realm, name string) (*FlowView, error)
- func (s *Service) RestartFlowInstance(ctx context.Context, realm string, realmID int16, name string) (*FlowView, error)
- func (s *Service) UpdateFlowConfig(ctx context.Context, realm, name string, config json.RawMessage) (*FlowView, error)
- func (s *Service) UpdatePipeline(ctx context.Context, realm, name string, definition []byte) (*PipelineView, error)
- func (s *Service) UpdateUserBlock(ctx context.Context, realm, name, blockType string, ...) (*UserBlockView, error)
- type UserBlockView
Constants ¶
This section is empty.
Variables ¶
var ErrValidation = errors.New("flowapi: validation failed")
ErrValidation marks a well-formed request that fails pipeline/block rules.
Functions ¶
This section is empty.
Types ¶
type API ¶
type API struct {
// contains filtered or unexported fields
}
API is the /flow/v1 HTTP surface. Routes accept a realm JWT carrying the upstream Flow claim a_f OR Astrate's original operator claim a_rma (any-of; issue #88).
type CreateFlowRequest ¶
type CreateFlowRequest struct {
Name string
Pipeline string
Config json.RawMessage
AutoRestart bool
}
CreateFlowRequest is the POST /flows body after defaults are applied.
type FlowView ¶
type FlowView struct {
Name string `json:"name"`
Pipeline string `json:"pipeline"`
Realm string `json:"realm"`
Config json.RawMessage `json:"config"`
AutoRestart bool `json:"auto_restart"`
Status string `json:"status"`
ErrorMessage *string `json:"error_message"`
FailedBlock *string `json:"failed_block,omitempty"`
RuntimeID string `json:"runtime_id,omitempty"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
StartedAt *time.Time `json:"started_at"`
StoppedAt *time.Time `json:"stopped_at"`
}
FlowView is the operator JSON shape for one durable flow (merged with live status when the instance is running in the Manager).
type PipelineView ¶
type PipelineView struct {
Name string `json:"name"`
Definition json.RawMessage `json:"definition"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
// ReferencingFlows is set on update: names of durable flows built from
// this pipeline. Running instances keep executing the old definition
// until explicitly reloaded (POST /flows/{name}/reload).
ReferencingFlows []string `json:"referencing_flows,omitempty"`
}
PipelineView is the operator JSON shape for one stored pipeline.
type Service ¶
type Service struct {
// contains filtered or unexported fields
}
Service implements pipeline CRUD and durable flow lifecycle over store + Manager.
func NewService ¶
func NewService(st *store.Store, mgr *flow.Manager, reg *flow.Registry, bus *stream.Bus, ingest flow.IngestFunc, register flow.RegisterFunc, log *slog.Logger) *Service
NewService wires store, manager, block registry, and the live bus. mgr and reg must be non-nil; bus may be nil only if no pipeline uses bus-backed sources; ingest may be nil only if no pipeline uses virtual_device_pool (#84); register may be nil only if no pipeline uses virtual_device_pool with auto_register.
func (*Service) CreateAndStartFlow ¶
func (s *Service) CreateAndStartFlow(ctx context.Context, realm string, req CreateFlowRequest) (*FlowView, error)
CreateAndStartFlow inserts a durable flow row and starts it (single start path).
func (*Service) CreatePipeline ¶
func (s *Service) CreatePipeline(ctx context.Context, realm, name string, definition []byte) (*PipelineView, error)
CreatePipeline validates and stores a pipeline definition.
func (*Service) CreateUserBlock ¶
func (s *Service) CreateUserBlock(ctx context.Context, realm, name, blockType string, source, configSchema json.RawMessage) (*UserBlockView, error)
CreateUserBlock validates and stores a user-defined composite block. The source must expand cleanly right now against built-ins plus the realm's stored blocks, so self-cycles and references to not-yet-existing composites are refused at creation time (accepted v1 behaviour; upstream defers the same failure to build time).
func (*Service) DeleteFlow ¶
DeleteFlow stops a live instance if present, unregisters it, and deletes the durable row.
func (*Service) DeletePipeline ¶
DeletePipeline removes a stored pipeline. Running flows are not stopped.
func (*Service) DeleteUserBlock ¶
DeleteUserBlock removes a stored user block. Built-in (default) blocks cannot be deleted through the user-block surface.
func (*Service) GetPipeline ¶
GetPipeline returns one pipeline by name.
func (*Service) GetUserBlock ¶
GetUserBlock returns one stored user block by name.
func (*Service) HasBuiltInBlock ¶
HasBuiltInBlock reports whether blockType names a registered built-in; GET /blocks/{name} dispatches on it.
func (*Service) ListBlocks ¶
ListBlocks returns operator docs for every registered block type. Realm is accepted for path symmetry with other Flow routes; the catalog is process-global.
func (*Service) ListFlows ¶
ListFlows returns durable flows for a realm (including stopped/failed), merged with live Manager status when running.
func (*Service) ListPipelines ¶
ListPipelines returns every pipeline name for a realm.
func (*Service) ListUserBlocks ¶
ListUserBlocks returns every stored user block of a realm, ordered by name.
func (*Service) MarkRunningFlowsStopped ¶
MarkRunningFlowsStopped best-effort sets durable status to stopped for instances that were running (clean process shutdown).
func (*Service) RehydrateAutoRestart ¶
RehydrateAutoRestart starts every durable flow with auto_restart=true. Individual failures are logged and recorded as status=failed; boot continues. Returns an error only if listing rows fails.
func (*Service) ReloadFlow ¶
ReloadFlow re-resolves a flow's pipeline by name and rebuilds the live graph from it (issue #44). Explicit: editing a pipeline never touches referencing flows on its own.
func (*Service) RestartFlowInstance ¶
func (s *Service) RestartFlowInstance(ctx context.Context, realm string, realmID int16, name string) (*FlowView, error)
RestartFlowInstance tears down the live graph (if any) and rebuilds it from the durable row's current pipeline + config. The durable row is never deleted; on rebuild failure the flow is marked failed with the error. Works for running flows (reload semantics) and stopped/failed ones (manual start).
func (*Service) UpdateFlowConfig ¶
func (s *Service) UpdateFlowConfig(ctx context.Context, realm, name string, config json.RawMessage) (*FlowView, error)
UpdateFlowConfig replaces a flow's config snapshot (issue #46). The new config must substitute cleanly against the flow's current pipeline; if the flow is live it is rebuilt immediately with the new config.
func (*Service) UpdatePipeline ¶
func (s *Service) UpdatePipeline(ctx context.Context, realm, name string, definition []byte) (*PipelineView, error)
UpdatePipeline replaces a pipeline definition. Running flows built from it keep executing the old definition; the response lists them so the edit is never a silent no-op (issue #44).
func (*Service) UpdateUserBlock ¶
func (s *Service) UpdateUserBlock(ctx context.Context, realm, name, blockType string, source, configSchema json.RawMessage) (*UserBlockView, error)
UpdateUserBlock replaces a user block's type, source and schema. Built-in (default) blocks cannot be modified through the user-block surface.
type UserBlockView ¶
type UserBlockView struct {
Name string `json:"name"`
BlockType string `json:"block_type"`
Source json.RawMessage `json:"source"`
ConfigSchema json.RawMessage `json:"config_schema,omitempty"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
}
UserBlockView is the operator JSON shape for one stored user-defined composite block (#85). ConfigSchema is omitted when the block has none.
Source Files