ekfa/docs/service_websocket_protocol.md

EFKA 与 Service WebSocket 交互协议

本文档描述部署在 efka 主机上的 service 与 efka 之间的 WebSocket 通讯逻辑。当前接入实现位于：

• efka_service_channel.erl
• service.proto
• efka_subscription.erl
• efka_service.erl

1. 连接入口

efka 启动时会创建 Cowboy WebSocket server，监听配置来自 efka.config 的 websocket_server：

{websocket_server, [
    {port, 18080},
    {acceptors, 10},
    {max_connections, 1024},
    {backlog, 1024}
]}

WebSocket 路由固定为：

GET /ws

连接建立后，每个 WebSocket 连接对应一个 efka_service_channel 进程。

当前协议没有独立的认证字段。service 必须先发送 register request 完成注册；未注册前只能处理 register，其它 request 会返回 invalid request，cast 会被忽略。

Cowboy WebSocket ping 帧由 efka_service_channel 直接回复 pong：

websocket_handle(ping, State) -> {reply, pong, State}

2. 二进制帧格式

service 与 efka 之间使用 WebSocket binary frame。每个业务 frame 的第 1 个字节是自定义帧类型，剩余字节是 protobuf payload。

<frame_type:1 byte><protobuf_payload:N bytes>

帧类型：

frame_type	十六进制	方向	语义	protobuf 类型
REQUEST	0x01	service -> efka	请求，需要 efka 回复	ServiceRequest
REPLY	0x02	efka -> service	request 的响应	ServiceReply
CAST	0x03	双向	单向消息，不要求回复	ServiceCast

efka_service_channel 当前只处理 service 发来的 REQUEST 和 CAST。收到未知格式时只记录日志，不主动发送错误响应。

3. Protobuf 定义

协议定义来自 apps/efka/proto/service.proto：

syntax = "proto3";

message ServiceRequest {
  uint32 packet_id = 1;

  message Register {
    string service_id = 1;
  }

  message Subscribe {
    string topic = 1;
  }

  oneof request {
    Register register = 10;
    Subscribe subscribe = 11;
  }
}

message ServiceReply {
  uint32 packet_id = 1;

  message Error {
    int32 code = 1;
    string message = 2;
  }

  oneof reply {
    bytes result = 10;
    Error error = 11;
  }
}

message ServiceCast {
  message MetricData {
    bytes route_key = 1;
    bytes metric = 2;
  }

  message TopicEvent {
    string topic = 1;
    bytes content = 2;
  }

  oneof body {
    TopicEvent topic_event = 10;
    MetricData metric_data = 11;
  }
}

注意：

• ServiceRequest.packet_id 由 service 生成，用于匹配 request 和 reply。
• ServiceReply.packet_id 必须等于原 request 的 packet_id。
• ServiceReply.result 当前成功值一般为 <<"ok">>。
• ServiceReply.error.code 当前失败时统一使用 -1。
• MetricData.route_key 和 MetricData.metric 是 bytes。
• TopicEvent.topic 是 string，TopicEvent.content 是 bytes。

4. request/reply 语义

4.1 register

service 建立 WebSocket 后应先发送注册请求：

ServiceRequest {
  packet_id: 1
  register {
    service_id: "service-a"
  }
}

WebSocket binary frame：

0x01 ++ protobuf(ServiceRequest)

efka 处理逻辑：

1. efka_service_channel 解码 ServiceRequest。
2. 调用 efka_service_sup:start_service(ServiceId) 启动或复用 service 进程。
3. 调用 efka_service:attach_channel(ServicePid, ChannelPid) 绑定当前 WebSocket channel。
4. 通过 efka_service_model:insert/1 写入 service 记录：

#service{
    service_id = ServiceId,
    container_name = <<>>,
    status = ?SERVICE_RUNNING,
    meta_data = #{},
    create_ts = efka_util:timestamp(),
    update_ts = efka_util:timestamp()
}

如果该 service_id 是新记录，status 会写入 ?SERVICE_RUNNING。如果该 service_id 已存在，当前 efka_service_model:insert/1 只更新 meta_data、container_name 和 update_ts，不会覆盖已有 status。

5. efka_service_channel 状态更新为：

#state{
    service_id = ServiceId,
    service_pid = ServicePid,
    is_registered = true
}

成功响应：

ServiceReply {
  packet_id: 1
  result: "ok"
}

WebSocket binary frame：

0x02 ++ protobuf(ServiceReply)

失败响应：

ServiceReply {
  packet_id: 1
  error {
    code: -1
    message: "attach channel failed"
  }
}

常见失败场景：

• 同一个 service_id 已有存活 channel，efka_service:attach_channel/2 返回 {error, <<"channel exists">>}。
• 当前实现对外统一返回 "attach channel failed"，详细原因只写日志。

4.2 subscribe

注册成功后，service 可以订阅 topic：

ServiceRequest {
  packet_id: 2
  subscribe {
    topic: "device/*/event"
  }
}

efka 处理逻辑：

1. 要求当前 channel 已注册，即 is_registered = true。
2. 调用 efka_subscription:subscribe(Topic, ChannelPid)。
3. 订阅成功后，把 topic 加入 efka_service_channel 的 subscribed_topics 集合。

成功响应：

ServiceReply {
  packet_id: 2
  result: "ok"
}

失败响应：

ServiceReply {
  packet_id: 2
  error {
    code: -1
    message: "invalid topic name"
  }
}

未注册时发送 subscribe 会走通用错误：

ServiceReply {
  packet_id: 2
  error {
    code: -1
    message: "invalid request"
  }
}

5. cast 语义

5.1 service -> efka: metric_data

service 通过 CAST 上报指标数据：

ServiceCast {
  metric_data {
    route_key: "device/a/temp"
    metric: "{\"value\":23.5}"
  }
}

WebSocket binary frame：

0x03 ++ protobuf(ServiceCast)

efka 处理逻辑：

1. efka_service_channel 解码 ServiceCast。
2. 要求当前 channel 已注册。
3. 调用：

efka_service:metric_data(ServicePid, RouteKey, Metric)

4. efka_service 再调用：

efka_iot_client:metric_data(RouteKey, Metric)

5. efka_iot_client 通过 efka 与 iot 的 TLS 长连接把数据上报给 iot：

{<<"message">>, {<<"data">>, #{
    <<"route_key">> => RouteKey,
    <<"metric">> => Metric
}}}

metric_data 是单向消息，service 不会收到 reply。未注册或未知 cast body 会被忽略。

5.2 efka -> service: topic_event

当 iot 通过 efka 与 iot 的 TLS channel 下发 pub 消息到 efka 时，efka_iot_client 会调用：

efka_subscription:publish(Topic, Qos, Content)

efka_subscription 根据 topic 匹配已订阅的 efka_service_channel。匹配成功后向 channel 进程发送：

{topic_broadcast, Topic, Content}

efka_service_channel 再推送 WebSocket CAST 给 service：

ServiceCast {
  topic_event {
    topic: "device/a/event"
    content: "payload bytes"
  }
}

WebSocket binary frame：

0x03 ++ protobuf(ServiceCast)

topic_event 是单向消息，不要求 service 回复。

6. topic 匹配规则

efka_subscription 当前使用 / 拆分 topic：

binary:split(Topic, <<$/>>, [global])

支持两种通配符：

通配符	语义
*	单级匹配，只匹配一个 segment
+	多级匹配，只能出现在最后一段，且至少匹配一个剩余 segment

示例：

订阅 topic	发布 topic	是否匹配
device/a/temp	device/a/temp	是
device/*/temp	device/a/temp	是
device/*/temp	device/a/b/temp	否
device/+	device/a	是
device/+	device/a/temp	是
device/+	device	否

同一个 efka_service_channel 订阅多个 topic 时，同一次 publish 最多投递一次。

7. QoS 与遗留消息

efka_subscription:publish(Topic, Qos, Content) 中的 Qos 影响没有订阅者时的行为：

• Qos = 0：没有匹配订阅者时直接丢弃。
• Qos /= 0：没有匹配订阅者时暂存到 remaining_messages。

当后续有 service 订阅能匹配这些 topic 时，efka_subscription 会把对应遗留消息推送给该 service，并从 remaining_messages 移除。

当前 remaining_messages 存在内存中，不持久化；efka 重启后会丢失。

8. 生命周期与清理

WebSocket channel 关闭

efka_service_channel:terminate/3 会执行：

1. 调用 efka_subscription:unsubscribe_all(ChannelPid) 清理该 channel 的所有订阅。
2. 如果 channel 已完成 register，则调用：

efka_service_model:change_status(ServiceId, ?SERVICE_STOPPED)

也就是把 service 状态更新为停止。

service 进程退出

efka_service_channel monitor 绑定的 efka_service 进程。如果 service 进程退出，channel 会停止：

websocket_info({'DOWN', _Ref, process, ServicePid, Reason}, State) ->
    {stop, State#state{service_pid = undefined}}

efka_service 同时也 monitor channel。channel 退出时，efka_service 会清空自身的 channel_pid。

9. 完整交互流程

典型流程：

service                         efka/efka_service_channel              efka_service              efka_subscription
  |                                      |                              |                            |
  |-- WebSocket connect /ws ------------>|                              |                            |
  |                                      |                              |                            |
  |-- REQUEST register ----------------->|                              |                            |
  |                                      |-- start_service(service_id) ->|                            |
  |                                      |-- attach_channel ----------->|                            |
  |<------------- REPLY result="ok" -----|                              |                            |
  |                                      |                              |                            |
  |-- REQUEST subscribe ---------------->|                              |                            |
  |                                      |---------------- subscribe ------------------------------>|
  |<------------- REPLY result="ok" -----|                              |                            |
  |                                      |                              |                            |
  |-- CAST metric_data ----------------->|                              |                            |
  |                                      |-- metric_data -------------->|                            |
  |                                      |                              |-- efka_iot_client:metric_data -> iot
  |                                      |                              |                            |
  |<------------- CAST topic_event ------|<--------------- topic_broadcast -----------------------|
  |                                      |                              |                            |
  |-- WebSocket close ------------------>|                              |                            |
  |                                      |-- unsubscribe_all -------------------------------------->|
  |                                      |-- change_status(service_id, stopped)

10. 当前限制

• 没有认证字段，service 只通过 service_id 注册。
• register 可重复连接同一个 service_id，但同一时间只允许一个活跃 channel 绑定到 efka_service。
• request 只支持 register 和 subscribe。
• cast 只支持 metric_data 和 topic_event。
• subscribed_topics 目前只在 efka_service_channel 内记录，关闭时实际清理由 efka_subscription:unsubscribe_all/1 完成。
• efka_subscription 当前基于列表扫描匹配，订阅规模很大时需要后续优化为索引结构。