feat: bootstrap LiveKit dev service and RocketMQ topics

Modified app/src/main.rs +38 −13 View app/src/main.rs

@@ -1,4 +1,4 @@

1#![recursion_limit = "256"]

1#![recursion_limit = "512"]

22

33mod cli;

44mod event;

@@ -64,6 +64,22 @@

6464 let conn = queue::RocketMqConn::init(&cfg.rocketmq, &mq_runtime)

6565 .await

6666 .map_err(|err| anyhow::anyhow!("failed to start the rocketmq producer: {err}"))?;

67 let topics = event_topics(&cfg.rocketmq);

68 let topic_report = conn.ensure_topics(&topics).await;

69 if topic_report.failed.is_empty() {

70 info!(

71 created = topic_report.created.len(),

72 existing = topic_report.existing.len(),

73 "event topics ready"

74 );

75 } else {

76 warn!(

77 created = topic_report.created.len(),

78 existing = topic_report.existing.len(),

79 failed = ?topic_report.failed,

80 "event topic ensure incomplete"

81 );

82 }

6783

6884 let auth_events = auth::events::EventHandler::new(conn.clone());

6985 let email_pool: config::email::EmailPoolConfig = config::loader::figment()

@@ -99,7 +115,7 @@

99115 ],

100116 );

101117

102 spawn_dependency_health_sampler(state.clone(), conn.clone());

118 spawn_dependency_health_sampler(state.clone(), conn.clone(), topics);

103119

104120 let mut workers = tokio::task::JoinSet::new();

105121 workers.spawn(chat::worker::run_background_tasks(

@@ -447,24 +463,33 @@

447463const HEALTH_SAMPLE_INTERVAL: std::time::Duration = std::time::Duration::from_secs(10);

448464const HEALTH_PROBE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(3);

449465

466/// 事件 topic 全集：`EventKind::ALL` 的 area 去重后按前缀拼装。

467fn event_topics(config: &config::rocketmq::RocketMqConfig) -> Vec<String> {

468 let mut areas: Vec<&'static str> = events::EventKind::ALL

469 .iter()

470 .map(|kind| kind.topic())

471 .collect();

472 areas.sort_unstable();

473 areas.dedup();

474 areas.into_iter().map(|area| config.topic(area)).collect()

475}

476

450477/// 周期性采样 RocketMQ/S3 健康：`/readyz` 只读缓存，避免每次探活都新建连接。

451fn spawn_dependency_health_sampler(state: bootstrap::AppState, conn: queue::RocketMqConn) {

478fn spawn_dependency_health_sampler(

479 state: bootstrap::AppState,

480 conn: queue::RocketMqConn,

481 topics: Vec<String>,

482) {

452483 tokio::spawn(async move {

453484 let mut ticker = tokio::time::interval(HEALTH_SAMPLE_INTERVAL);

454485 ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);

455 let auth_topic = state

456 .config

457 .rocketmq

458 .topic(events::EventKind::AuthLoginCreate.topic());

459486 loop {

460487 ticker.tick().await;

461488

462 let rocketmq_up = tokio::time::timeout(

463 std::time::Duration::from_secs(6),

464 conn.health(&[&auth_topic, "TBW102"]),

465 )

466 .await

467 .is_ok_and(|result| result.is_ok());

489 let rocketmq_up =

490 tokio::time::timeout(std::time::Duration::from_secs(6), conn.health(&topics))

491 .await

492 .is_ok_and(|result| result.is_ok());

468493 state.health.mark_rocketmq(rocketmq_up);

469494 oltp::metrics::health_dependency("rocketmq", rocketmq_up);

470495

Added deploy/livekit/livekit.yaml +51 −0 View deploy/livekit/livekit.yaml

@@ -0,0 +1,51 @@

1# KenApi 本地开发用 LiveKit Server 配置。

2#

3# 与 config/local/livekit.json 一一对应：

4# url = ws://127.0.0.1:7880 -> port: 7880

5# api_url = http://127.0.0.1:7880 -> 同一进程

6# api_key / api_secret -> keys 下的唯一一对

7# empty_timeout_secs / departure_timeout_secs / default_max_participants -> room

8#

9# webhook 指向 apihost 的公开回调路由（lib/service/webhook: /chat/livekit/webhook

10# 挂在 /api/v1 下），由 LiveKit 用 api_key 对应的 secret 签名，服务端用

11# TokenVerifier 校验，因此 api_key 必须与 keys 里的一致。

12#

13# 容器以 host 网络运行（与 deploy/compose.yaml 里的 apihost 一致），所以这里的

14# 127.0.0.1 就是宿主机，Redis 与 apihost 都不需要额外的服务名解析。

15

16port: 7880

17

18rtc:

19 # WebRTC over TCP 回落端口（浏览器 UDP 不通时走这里）。

20 tcp_port: 7881

21 # 单端口 UDP mux，避免在 host 网络上开放整段端口范围。

22 udp_port: 7882

23 # 本机/局域网直连，不做公网 IP 探测；如需对外暴露改成 true 并配置 node_ip。

24 use_external_ip: false

25

26redis:

27 address: 127.0.0.1:6379

28 # 与 apihost 的 redis db 分开，避免 key 前缀互相污染。

29 db: 1

30 use_tls: false

31

32keys:

33 APIe7BAcijKR87b: nO8qfPxbAWkIfKNbxhsan2gVnR1YK0JfXiWUJd1EL1fA

34

35webhook:

36 api_key: APIe7BAcijKR87b

37 urls:

38 - http://127.0.0.1:7051/api/v1/chat/livekit/webhook

39

40room:

41 empty_timeout: 300

42 departure_timeout: 20

43 max_participants: 50

44

45turn:

46 # 本地开发不需要 TURN；UDP mux + TCP 回落已覆盖同机与局域网。

47 enabled: false

48

49logging:

50 level: info

51 json: false

Modified lib/infra/queue/src/conn.rs +65 −11 View lib/infra/queue/src/conn.rs

@@ -1,3 +1,6 @@

1use std::collections::HashMap;

2use std::time::Duration;

3

14use config::rocketmq::RocketMqConfig;

25use rocketmq_client_rust::DefaultMQProducer;

36use rocketmq_model::common::message::message_single::Message;

@@ -9,6 +12,20 @@

912/// 事件信封的字节预算：为消息属性与协议头预留 4 KiB。

1013pub const MAX_ENVELOPE_BYTES: usize = DEFAULT_MAX_MESSAGE_BYTES - 4096;

1114

15/// 建 topic 时向 broker 提交的模板 topic，与 `TopicValidator::AUTO_CREATE_TOPIC_KEY_TOPIC` 一致。

16const CREATE_TOPIC_KEY: &str = "TBW102";

17const DEFAULT_TOPIC_QUEUE_NUMS: i32 = 4;

18const ENSURE_ROUTE_TIMEOUT: Duration = Duration::from_secs(3);

19const ENSURE_CREATE_TIMEOUT: Duration = Duration::from_secs(5);

20

21/// `ensure_topics` 的结果：补齐的 topic、已存在的 topic 与失败明细。

22#[derive(Debug, Default, Clone)]

23pub struct TopicEnsureReport {

24 pub created: Vec<String>,

25 pub existing: Vec<String>,

26 pub failed: Vec<(String, String)>,

27}

28

1229#[derive(Clone)]

1330pub struct RocketMqConn {

1431 producer: DefaultMQProducer,

@@ -59,21 +76,58 @@

5976 self.config.max_message_bytes

6077 }

6178

62 pub async fn health(&self, topics: &[&str]) -> QueueResult<()> {

79 /// 全部 topic 的路由都能查到才算健康：任一 topic 缺失说明事件通道不完整。

80 pub async fn health(&self, topics: &[String]) -> QueueResult<()> {

81 let mut producer = self.producer.clone();

82 let mut failures = Vec::new();

83 for topic in topics {

84 if let Err(error) = producer.fetch_publish_message_queues(topic).await {

85 failures.push((topic, error));

86 }

87 }

88 match failures.first() {

89 None => Ok(()),

90 Some((topic, error)) => Err(QueueError::connect(format!(

91 "rocketmq route lookup failed for {}/{} topics (first: `{topic}`: {error})",

92 failures.len(),

93 topics.len()

94 ))),

95 }

96 }

97

98 /// 幂等补齐事件 topic：路由存在即跳过，缺失才向 broker 提交创建请求。

99 pub async fn ensure_topics(&self, topics: &[String]) -> TopicEnsureReport {

63100 let mut producer = self.producer.clone();

64 let mut last_error = None;

101 let mut report = TopicEnsureReport::default();

65102 for topic in topics {

66 match producer.fetch_publish_message_queues(topic).await {

67 Ok(_) => return Ok(()),

68 Err(error) => last_error = Some((*topic, error)),

103 let route = tokio::time::timeout(

104 ENSURE_ROUTE_TIMEOUT,

105 producer.fetch_publish_message_queues(topic),

106 )

107 .await;

108 if matches!(route, Ok(Ok(_))) {

109 report.existing.push(topic.clone());

110 continue;

111 }

112 let created = tokio::time::timeout(

113 ENSURE_CREATE_TIMEOUT,

114 producer.create_topic(

115 CREATE_TOPIC_KEY,

116 topic,

117 DEFAULT_TOPIC_QUEUE_NUMS,

118 HashMap::new(),

119 ),

120 )

121 .await;

122 match created {

123 Ok(Ok(())) => report.created.push(topic.clone()),

124 Ok(Err(error)) => report.failed.push((topic.clone(), error.to_string())),

125 Err(_) => report

126 .failed

127 .push((topic.clone(), "create topic timed out".to_owned())),

69128 }

70129 }

71 let detail = last_error

72 .map(|(topic, error)| format!("`{topic}`: {error}"))

73 .unwrap_or_default();

74 Err(QueueError::connect(format!(

75 "rocketmq route lookup failed ({detail})"

76 )))

130 report

77131 }

78132

79133 pub async fn send(&self, topic: &str, message: Message) -> QueueResult<()> {

Modified lib/infra/queue/src/lib.rs +1 −1 View lib/infra/queue/src/lib.rs

@@ -8,7 +8,7 @@

88pub mod runtime;

99pub mod subject;

1010

11pub use conn::{DEFAULT_MAX_MESSAGE_BYTES, MAX_ENVELOPE_BYTES, RocketMqConn};

11pub use conn::{DEFAULT_MAX_MESSAGE_BYTES, MAX_ENVELOPE_BYTES, RocketMqConn, TopicEnsureReport};

1212pub use consumer::{Consumer, ConsumerConfig, MessageHandler};

1313pub use envelope::{DeliveryClass, EVENT_PREFIX, Envelope, Labels, SCHEMA_VERSION};

1414pub use error::{QueueError, QueueResult};

Modified scripts/dev.sh +1 −1 View scripts/dev.sh

@@ -28,7 +28,7 @@

2828

2929ORDER="apihost"

3030BACKENDS="apihost"

31INFRA="postgresql:5432 redis:6379 s3:9000"

31INFRA="postgresql:5432 redis:6379 s3:9000 livekit:7880"

3232

3333if [ -t 1 ]; then

3434 C_WARN="$(printf '\033[33m')"; C_ERR="$(printf '\033[31m')"; C_OFF="$(printf '\033[0m')"

Modified docker-compose.dev.yml +20 −1 View docker-compose.dev.yml

@@ -1,6 +1,6 @@

11# 本地开发依赖服务（macOS / Linux）。

22# 与 config/local/ 的默认端口一一对应：

3# PostgreSQL 5432 / Redis 6379 / S3(RustFS) 9000

3# PostgreSQL 5432 / Redis 6379 / S3(RustFS) 9000 / LiveKit 7880(7881/7882)

44# RocketMQ 是 KenApi 的启动必需项，由宿主机提供：

55# config/local/rocketmq.json 覆盖为 127.0.0.1:6876，未建 local 时用

66# config/default/rocketmq.json 的 127.0.0.1:9876。

@@ -66,6 +66,25 @@

6666 timeout: 3s

6767 retries: 20

6868

69 # 语音/视频房间服务（config/local/livekit.json 指向 127.0.0.1:7880）。

70 # 用 host 网络：LiveKit 既要读宿主机 Redis（127.0.0.1:6379），又要 webhook 回调

71 # 宿主机 apihost（127.0.0.1:7051/api/v1/chat/livekit/webhook），host 网络省掉

72 # 服务名解析与端口映射。配置见 deploy/livekit/livekit.yaml，key/secret 与

73 # config/local/livekit.json 一致。

74 # 注意：host 网络只在 Linux 生效；macOS/Windows 需改桥接并设 rtc.node_ip。

75 livekit:

76 image: ${LIVEKIT_IMAGE:-docker.m.daocloud.io/livekit/livekit-server:v1.9.1}

77 container_name: gitdata-dev-livekit

78 restart: unless-stopped

79 network_mode: host

80 command: ["--config", "/etc/livekit/livekit.yaml"]

81 volumes:

82 - ./deploy/livekit/livekit.yaml:/etc/livekit/livekit.yaml:ro

83 logging:

84 driver: json-file

85 options:

86 max-size: "20m"

87 max-file: "3"

6988

7089 # 本地 S3 bucket 初始化（建出 config/local/s3.json 里声明的 bucket）

7190 minio-init: