Godot SDK — connect your game in minutes

## FlowFn Streams client for Godot 4.x — protocol v1.
##
## One node wraps the whole wire protocol: connect with a signed player
## token (production) or a bare user id (open-identity prototyping), join
## rooms, publish JSON on channels, and react to signals. Add it as an
## autoload (e.g. `FlowFnStream`) or instance it per scene; it polls itself
## in _process, no manual poll loop needed.
##
## Quickstart:
##     var fs := FlowFnStream.new()
##     add_child(fs)
##     fs.message.connect(_on_stream_message)
##     fs.joined.connect(func(room, members, _ref): print("in ", room, " with ", members.size()))
##     fs.connect_with_token("strm-7f3a2c1d", token_from_your_server)
##     await fs.connected
##     fs.join_room("match_8f3a")                      # every channel
##     fs.join_room("spectate_1", ["events", "ai"])    # selective subscription
##     fs.publish("match_8f3a", "chat", {"text": "gg"})
##
## Tokens: your game server mints them server-side —
##     POST https://stream.flowfn.com/stream/v1/tokens
##     Authorization: Bearer <stream server key>
##     {"stream": "strm-…", "user_id": "p1", "name": "Ana", "rooms": ["match_8f3a"]}
## Never ship the server key inside a client build.
class_name FlowFnStream
extends Node

## Emitted after the gateway's `welcome` frame — safe to join rooms.
signal connected(channels: Array)
## A `join` acked: members is the room roster including you.
signal joined(room: String, members: Array, ref: String)
## A `leave` acked.
signal left(room: String, ref: String)
signal peer_joined(room: String, user: Dictionary)
signal peer_left(room: String, user: Dictionary)
## Every message you are subscribed to: from = {user_id, name, role}.
## role is "player", "server", or "flowfn" (engine-side: bindings, AI NPCs,
## workflow results — players can never spoof it).
signal message(room: String, channel: String, from: Dictionary, payload: Variant, ts: int)
## Protocol error frames: code is a stable string (rate_limited, room_full…).
signal stream_error(code: String, message: String, ref: String)
## Socket closed. 4001 unauthorized · 4003 stream disabled · 4004 connection
## limit · 4008 rate-limit kick · 4009 server restarting (just reconnect).
signal disconnected(code: int, reason: String)

## Gateway base URL — override for self-hosted/staging.
@export var gateway_url: String = "wss://stream.flowfn.com/stream/v1"
## Reconnect automatically with backoff after an unexpected close.
@export var auto_reconnect: bool = true

var _socket: WebSocketPeer = WebSocketPeer.new()
var _url: String = ""
var _state: int = WebSocketPeer.STATE_CLOSED
var _was_connected := false
var _reconnect_wait := 1.0
var _reconnect_timer := 0.0
var _rejoin: Dictionary = {}   # room -> channels filter (Array or null)
var _ref_seq := 0

## Production: connect with a signed player token minted by YOUR server.
func connect_with_token(stream_code: String, token: String) -> void:
	_open("%s?stream=%s&token=%s" % [gateway_url, stream_code.uri_encode(), token.uri_encode()])

## Prototyping only: requires the stream's "open identity" toggle.
func connect_open(stream_code: String, user_id: String, display_name: String = "") -> void:
	var url := "%s?stream=%s&user_id=%s" % [gateway_url, stream_code.uri_encode(), user_id.uri_encode()]
	if display_name != "":
		url += "&name=%s" % display_name.uri_encode()
	_open(url)

## Trusted game-server process only — NEVER ship the key in a client build.
func connect_with_server_key(stream_code: String, server_key: String) -> void:
	_open("%s?stream=%s&key=%s" % [gateway_url, stream_code.uri_encode(), server_key.uri_encode()])

## Join a room. Pass `channels` to subscribe to a subset (saves bandwidth);
## empty array = every channel. Rooms are created by joining them.
func join_room(room: String, channels: Array = [], ref: String = "") -> void:
	var frame := {"op": "join", "room": room}
	if channels.size() > 0:
		frame["channels"] = channels
	_rejoin[room] = channels if channels.size() > 0 else null
	_send(frame, ref)

func leave_room(room: String, ref: String = "") -> void:
	_rejoin.erase(room)
	_send({"op": "leave", "room": room}, ref)

## Publish a JSON-serializable payload on a channel (16 KB frame cap).
func publish(room: String, channel: String, payload: Variant, ref: String = "") -> void:
	_send({"op": "publish", "room": room, "channel": channel, "payload": payload}, ref)

func close() -> void:
	auto_reconnect = false
	_rejoin.clear()
	_socket.close(1000, "client closing")

func is_socket_connected() -> bool:
	return _state == WebSocketPeer.STATE_OPEN and _was_connected

func _open(url: String) -> void:
	_url = url
	_was_connected = false
	_socket = WebSocketPeer.new()
	var err := _socket.connect_to_url(url)
	if err != OK:
		push_error("FlowFnStream: connect_to_url failed (%d)" % err)
	set_process(true)

func _send(frame: Dictionary, ref: String) -> void:
	if _state != WebSocketPeer.STATE_OPEN:
		push_warning("FlowFnStream: not connected — dropped op %s" % str(frame.get("op")))
		return
	if ref == "":
		_ref_seq += 1
		ref = "r%d" % _ref_seq
	frame["ref"] = ref
	_socket.send_text(JSON.stringify(frame))

func _process(delta: float) -> void:
	if _url == "":
		return
	_socket.poll()
	var next_state := _socket.get_ready_state()

	if next_state == WebSocketPeer.STATE_OPEN:
		while _socket.get_available_packet_count() > 0:
			_handle_frame(_socket.get_packet().get_string_from_utf8())
	elif next_state == WebSocketPeer.STATE_CLOSED and _state != WebSocketPeer.STATE_CLOSED:
		var code := _socket.get_close_code()
		var reason := _socket.get_close_reason()
		disconnected.emit(code, reason)
		# 4001/4003 are deliberate refusals — reconnecting would just loop.
		if auto_reconnect and code != 4001 and code != 4003:
			_reconnect_timer = _reconnect_wait
			_reconnect_wait = minf(_reconnect_wait * 2.0, 30.0)
	_state = next_state

	if _state == WebSocketPeer.STATE_CLOSED and _reconnect_timer > 0.0:
		_reconnect_timer -= delta
		if _reconnect_timer <= 0.0:
			_open(_url)

func _handle_frame(text: String) -> void:
	var data: Variant = JSON.parse_string(text)
	if typeof(data) != TYPE_DICTIONARY:
		return
	var frame: Dictionary = data
	match str(frame.get("op", "")):
		"welcome":
			_was_connected = true
			_reconnect_wait = 1.0
			connected.emit(frame.get("channels", []))
			# Resume room membership after a reconnect.
			for room in _rejoin.keys():
				var channels: Variant = _rejoin[room]
				var f := {"op": "join", "room": room}
				if channels != null:
					f["channels"] = channels
				_send(f, "")
		"ack":
			var room := str(frame.get("room", ""))
			if frame.has("members"):
				joined.emit(room, frame.get("members", []), str(frame.get("ref", "")))
			elif room != "":
				left.emit(room, str(frame.get("ref", "")))
		"message":
			message.emit(
				str(frame.get("room", "")),
				str(frame.get("channel", "")),
				frame.get("from", {}),
				frame.get("payload"),
				int(frame.get("ts", 0)),
			)
		"peer_joined":
			peer_joined.emit(str(frame.get("room", "")), frame.get("user", {}))
		"peer_left":
			peer_left.emit(str(frame.get("room", "")), frame.get("user", {}))
		"error":
			stream_error.emit(str(frame.get("code", "")), str(frame.get("message", "")), str(frame.get("ref", "")))
		"pong":
			pass