# EFKA 与 Service WebSocket 交互协议

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

- [efka_service_channel.erl](/usr/local/code/cloudkit/efka/apps/efka/src/service/efka_service_channel.erl)
- [service.proto](/usr/local/code/cloudkit/efka/apps/efka/proto/service.proto)
- [efka_subscription.erl](/usr/local/code/cloudkit/efka/apps/efka/src/service/efka_subscription.erl)
- [efka_service.erl](/usr/local/code/cloudkit/efka/apps/efka/src/service/efka_service.erl)

## 1. 连接入口

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

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

WebSocket 路由固定为：

```http
GET /ws
```

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

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

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

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

## 2. 二进制帧格式

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

```text
<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`：

```protobuf
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 后应先发送注册请求：

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

WebSocket binary frame：

```text
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 记录：

```erlang
#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` 状态更新为：

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

成功响应：

```protobuf
ServiceReply {
  packet_id: 1
  result: "ok"
}
```

WebSocket binary frame：

```text
0x02 ++ protobuf(ServiceReply)
```

失败响应：

```protobuf
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：

```protobuf
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` 集合。

成功响应：

```protobuf
ServiceReply {
  packet_id: 2
  result: "ok"
}
```

失败响应：

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

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

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

## 5. cast 语义

### 5.1 service -> efka: metric_data

service 通过 `CAST` 上报指标数据：

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

WebSocket binary frame：

```text
0x03 ++ protobuf(ServiceCast)
```

`efka` 处理逻辑：

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

```erlang
efka_service:metric_data(ServicePid, RouteKey, Metric)
```

4. `efka_service` 再调用：

```erlang
efka_iot_client:metric_data(RouteKey, Metric)
```

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

```erlang
{<<"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` 会调用：

```erlang
efka_subscription:publish(Topic, Qos, Content)
```

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

```erlang
{topic_broadcast, Topic, Content}
```

`efka_service_channel` 再推送 WebSocket `CAST` 给 service：

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

WebSocket binary frame：

```text
0x03 ++ protobuf(ServiceCast)
```

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

## 6. topic 匹配规则

`efka_subscription` 当前使用 `/` 拆分 topic：

```erlang
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，则调用：

```erlang
efka_service_model:change_status(ServiceId, ?SERVICE_STOPPED)
```

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

### service 进程退出

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

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

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

## 9. 完整交互流程

典型流程：

```text
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` 当前基于列表扫描匹配，订阅规模很大时需要后续优化为索引结构。