flow_lab — local end-to-end flow dummy

Local-only Triggers SDK service for exercising the full v2 path:

• manifest registration against triggers-v2
• Kafka source → orchestrator walk → NATS service-in → action handler
• auto-derived result_event metadata for Studio chaining (parse → wrap)
• optional output-topic leg after wrap

Do not deploy. Service name flow_lab; the example code marks it with triggers.LifecycleBeta. Backend manifest lifecycle values remain stable, beta, and deprecated.

Actions

Action	Input	Output	result_event
parse	{ "message": "hello" }	{ "message", "length" }	flow_lab.parse.result
wrap	{ "message", "prefix"? }	{ "wrapped" }	flow_lab.wrap.result

NATS service-in subject: triggers.v2.flow_lab.inbound

Quick start

1. Start local v2 stack

From notify_triggers/:

make local-infra      # mongo + kafka on :9092
make v2-shared-up     # nats :4222, triggers-v2 API :8888

2. Register the Kafka source

From triggers-sdk-go/examples/flow_lab/:

make seed
# or: TRIGGERS_V2_API=http://localhost:8888 ./seed.sh

3. Run the dummy service

make run
# or: TRIGGERS_NATS_URL=nats://127.0.0.1:4222 go run .

Confirm registration:

curl -s http://localhost:8888/api/v2/services/flow_lab | jq '.liveness, [.manifest.actions[].name]'

Expect online and ["parse","wrap"].

4. Build a flow in Studio

Studio graph grammar (enforced by flows.Validate):

• Source never connects directly to an action — always route through a service node first.
• Action never connects directly to another action — exit via output-topic or another service, then enter the next service segment.
• In the canvas, actions are nested under their parent service.

Simple (source → parse):

source (flow_lab_in)
  → service (flow_lab · nats → triggers.v2.flow_lab.inbound)
      └── action (parse)

Map the source payload field message (or the whole body) into parse.message.

Chained (parse → wrap):

Two service segments — wrap receives the upstream bound fields from parse plus any literal overrides you set in the flow:

source (flow_lab_in)
  → service (flow_lab)
      └── action (parse)
  → service (flow_lab)
      └── action (wrap)     ← message ← parse.result.message; prefix ← "test"

On wrap, bind:

• message ← parse result field message
• prefix ← literal "test" (optional)

With output topic:

source (flow_lab_in)
  → service (flow_lab) · parse
  → service (flow_lab) · wrap
  → output-topic (flow_lab.out)

For flat typed struct outputs like the parse and wrap results here, the Go SDK auto-derives each action's result_event. That gives Studio the upstream fields it needs for chaining, while still letting you add literal overrides such as prefix = "test" on the downstream action. Empty outputs omit result_event. Use WithoutResultEvent() to opt out, or WithResultEvent(...) to override/customize the derived value explicitly when your output shape is not derivable.

Release the flow, then enable it if your Studio workflow requires that separately.

5. Trigger a run

After the flow is released against source flow_lab_in:

make produce
# or: echo '{"message":"hello"}' | rpk topic produce flow_lab.in --brokers 127.0.0.1:9092

Inspect runs:

curl -s "http://localhost:8888/api/v2/flows/<flow_id>/runs" | jq .

Scale-safe player casino source

The Python examples/11_flow_lab/ producer can exercise the RisingWave player-casino contract through the same runtime:

cd triggers-sdk-python/examples/11_flow_lab
make produce-casino

It publishes one record to rw-player-first-casino-state. RisingWave discovery exposes that source as event type rw_player_first_casino_state; the generated source ID is environment-specific and must not be hard-coded.

The money fields are fixed-scale USD integers:

• bet_amount_usd_pip
• win_amount_usd_pip
• deposit_amount_usd_pip
• amount_scale (100000000)
• amount_currency (USD)

Formula consumers must divide pip amounts by amount_scale before applying major-unit percentages or thresholds. Do not use the legacy bet_amount, win_amount, or deposit_amount fields for cross-currency rewards.

API bootstrap (optional)

Create + release a minimal chain without Studio:

API=http://localhost:8888

curl -sf -X POST "$API/api/v2/flows" \
  -H 'Content-Type: application/json' \
  -d '{"id":"flow_lab_chain","name":"Flow Lab chain","source_id":"flow_lab_in","status":"draft"}'

curl -sf -X POST "$API/api/v2/flows/flow_lab_chain/versions" \
  -H 'Content-Type: application/json' \
  -d '{
    "state":"saved",
    "body":{
      "id":"flow_lab_chain",
      "version":"v1",
      "nodes":[
        {"id":"src","kind":"source","name":"flow_lab_in"},
        {"id":"sn1","kind":"service-in","service":"flow_lab","config":{"transport":"nats","target":"triggers.v2.flow_lab.inbound"}},
        {"id":"parse","kind":"action","service":"sn1","name":"parse"},
        {"id":"sn2","kind":"service-in","service":"flow_lab","config":{"transport":"nats","target":"triggers.v2.flow_lab.inbound"}},
        {"id":"wrap","kind":"action","service":"sn2","name":"wrap"}
      ],
      "edges":[
        {"from":"src","to":"sn1"},
        {"from":"sn1","to":"parse"},
        {"from":"parse","to":"sn2"},
        {"from":"sn2","to":"wrap"}
      ]
    }
  }'

curl -sf -X POST "$API/api/v2/flows/flow_lab_chain/versions/1/release"

Wire wrap input bindings in Studio if release validation requires them, then make produce.

Harness test (no broker)

make test

Env vars

Var	Default	Purpose
TRIGGERS_NATS_URL	(required at runtime)	NATS for registration + dispatch
TRIGGERS_V2_API	http://localhost:8888	seed.sh target
KAFKA_BROKERS	127.0.0.1:9092	produce.sh, produce-casino
FLOW_LAB_TOPIC	flow_lab.in	produce.sh
PLAYER_FIRST_CASINO_TOPIC	rw-player-first-casino-state	produce-casino