feat(commits): enhance commits pagination with has_more and diff_view_style fields

Modified lib/infra/bootstrap/src/health.rs +59 −13 View lib/infra/bootstrap/src/health.rs

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

11use std::sync::Arc;

22use std::sync::atomic::{AtomicU8, Ordering};

3use std::time::Duration;

34

45use actix_web::HttpResponse;

56use sea_orm::{ConnectionTrait, DbBackend, Statement};

@@ -12,6 +13,10 @@

1213

1314const UP: u8 = 1;

1415const DOWN: u8 = 2;

16/// 连续失败达到该次数才置 DOWN：单次采样抖动不应把 Pod 摘出负载均衡。

17const DOWN_AFTER_FAILURES: u8 = 2;

18/// 探针自身超时，须小于 k8s `timeoutSeconds`（3s），否则超时也会被计为失败。

19const READY_PROBE_TIMEOUT: Duration = Duration::from_secs(2);

1520

1621/// 后台采样得到的依赖健康状态；探针只读取这里的值，不发起探测。

1722/// 原子量初值为 0，表示尚未采样（同样按未就绪处理）。

@@ -23,20 +28,18 @@

2328#[derive(Default)]

2429struct HealthCacheInner {

2530 rocketmq: AtomicU8,

31 rocketmq_failures: AtomicU8,

2632 s3: AtomicU8,

33 s3_failures: AtomicU8,

2734}

2835

2936impl HealthCache {

3037 pub fn mark_rocketmq(&self, up: bool) {

31 self.inner

32 .rocketmq

33 .store(if up { UP } else { DOWN }, Ordering::Relaxed);

38 mark(&self.inner.rocketmq, &self.inner.rocketmq_failures, up);

3439 }

3540

3641 pub fn mark_s3(&self, up: bool) {

37 self.inner

38 .s3

39 .store(if up { UP } else { DOWN }, Ordering::Relaxed);

42 mark(&self.inner.s3, &self.inner.s3_failures, up);

4043 }

4144

4245 fn dependencies(&self) -> [(&'static str, bool); 2] {

@@ -50,6 +53,19 @@

5053 }

5154}

5255

56/// 成功即置 UP；失败累计到阈值才置 DOWN，避免单次抖动摘流量。

57fn mark(state: &AtomicU8, failures: &AtomicU8, up: bool) {

58 if up {

59 failures.store(0, Ordering::Relaxed);

60 state.store(UP, Ordering::Relaxed);

61 return;

62 }

63 let count = failures.fetch_add(1, Ordering::Relaxed).saturating_add(1);

64 if count >= DOWN_AFTER_FAILURES {

65 state.store(DOWN, Ordering::Relaxed);

66 }

67}

68

5369#[derive(Clone)]

5470struct HealthState {

5571 database: Option<Arc<AppDatabase>>,

@@ -64,17 +80,20 @@

6480async fn readyz(State(state): State<HealthState>) -> HttpResponse {

6581 let db_ok = match state.database.as_ref() {

6682 None => true,

67 Some(db) => db

68 .write()

69 .query_one_raw(Statement::from_sql_and_values(

83 Some(db) => actix_web::rt::time::timeout(

84 READY_PROBE_TIMEOUT,

85 db.write().query_one_raw(Statement::from_sql_and_values(

7086 DbBackend::Postgres,

7187 "SELECT 1",

7288 [],

73 ))

74 .await

75 .is_ok(),

89 )),

90 )

91 .await

92 .is_ok_and(|result| result.is_ok()),

7693 };

77 let redis_ok = state.redis.ping().await.is_ok();

94 let redis_ok = actix_web::rt::time::timeout(READY_PROBE_TIMEOUT, state.redis.ping())

95 .await

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

7897

7998 let mut failed = Vec::new();

8099 if !db_ok {

@@ -118,3 +137,30 @@

118137 state.health.clone(),

119138 )

120139}

140

141#[cfg(test)]

142mod tests {

143 use super::*;

144

145 #[test]

146 fn down_needs_consecutive_failures_and_recovers_immediately() {

147 let state = AtomicU8::new(0);

148 let failures = AtomicU8::new(0);

149

150 // 尚未采样 + 首次失败：仍保持「未就绪」而不立刻判 DOWN。

151 mark(&state, &failures, false);

152 assert_eq!(state.load(Ordering::Relaxed), 0);

153

154 mark(&state, &failures, false);

155 assert_eq!(state.load(Ordering::Relaxed), DOWN);

156

157 // 一次成功立刻恢复，并清零失败计数。

158 mark(&state, &failures, true);

159 assert_eq!(state.load(Ordering::Relaxed), UP);

160 assert_eq!(failures.load(Ordering::Relaxed), 0);

161

162 // 恢复后再次单次失败仍然是 UP（阈值重新累计）。

163 mark(&state, &failures, false);

164 assert_eq!(state.load(Ordering::Relaxed), UP);

165 }

166}

Modified lib/service/chat/src/handlers/messages.rs +2 −2 View lib/service/chat/src/handlers/messages.rs

@@ -39,10 +39,10 @@

3939 get,

4040 path = "/chat/rooms/{room_id}/messages",

4141 tag = "chat",

42 description = "按房间序号过滤并分页读取消息；has_more 可用 offset + items.len() < total 判断。",

42 description = "按房间序号过滤并分页读取消息；items 按 seq 升序，附带 has_more / oldest_seq / newest_seq 供游标翻页。",

4343 params(("room_id" = Uuid, Path, description = "room id"), ListMessageQuery),

4444 responses(

45 (status = 200, description = "message page", body = AppResponseData<api::pagination::Page<MessageDto>>),

45 (status = 200, description = "message page", body = AppResponseData<MessagePage>),

4646 (status = 404, description = "not found", body = AppResponseData<NoData>),

4747 (status = 401, description = "authentication required", body = AppResponseData<NoData>)

4848 )

Modified lib/service/chat/src/services/messages.rs +31 −1 View lib/service/chat/src/services/messages.rs

@@ -133,7 +133,37 @@

133133

134134pub use social::MessageEmbed as MessageEmbedDto;

135135

136pub type MessagePage = api::pagination::Page<MessageDto>;

136/// 消息页：在统一分页字段之上补回游标翻页所需的派生字段。

137///

138/// `has_more`/`oldest_seq`/`newest_seq` 由本页内容推导，客户端无需自己算；

139/// `items` 恒按 `seq` 升序，因此 `oldest_seq` 取首条、`newest_seq` 取末条。

140#[derive(Debug, Clone, Serialize, ToSchema)]

141pub struct MessagePage {

142 pub items: Vec<MessageDto>,

143 pub total: u64,

144 pub limit: u64,

145 pub offset: u64,

146 pub has_more: bool,

147 pub oldest_seq: Option<i64>,

148 pub newest_seq: Option<i64>,

149}

150

151impl MessagePage {

152 pub fn new(items: Vec<MessageDto>, total: u64, limit: u64, offset: u64) -> Self {

153 let has_more = offset.saturating_add(items.len() as u64) < total;

154 let oldest_seq = items.first().map(|message| message.seq);

155 let newest_seq = items.last().map(|message| message.seq);

156 Self {

157 items,

158 total,

159 limit,

160 offset,

161 has_more,

162 oldest_seq,

163 newest_seq,

164 }

165 }

166}

137167

138168#[derive(Debug, Clone, Deserialize, Serialize, ToSchema)]

139169pub struct SendMessageCmd {

Modified lib/service/chat/src/share/fanout.rs +20 −1 View lib/service/chat/src/share/fanout.rs

@@ -6,7 +6,7 @@

66

77use std::sync::{Arc, OnceLock};

88

9use tokio::sync::{Mutex, mpsc};

9use tokio::sync::{Mutex, Semaphore, mpsc};

1010use tokio_util::sync::CancellationToken;

1111use tracing::Instrument as _;

1212use uuid::Uuid;

@@ -25,6 +25,9 @@

2525const BULK_CAPACITY: usize = 64;

2626/// @everyone 单块人数：一块的数据库往返次数有界，块与块之间让出 worker。

2727const EVERYONE_CHUNK: usize = 100;

28/// 满队列时最多允许多少个后台等待任务：这是 deferral 的硬上限，

29/// 保证持续过载时内存有界（超出即丢弃并打点，而不是无界堆积）。

30const MAX_DEFERRED: usize = 256;

2831

2932pub(crate) enum Job {

3033 MentionEmails {

@@ -119,6 +122,12 @@

119122}

120123

121124static FANOUT: OnceLock<Fanout> = OnceLock::new();

125/// 满队列转后台等待时占用的名额；名额耗尽说明已经过载，直接丢弃并打点。

126static DEFERRED: OnceLock<Arc<Semaphore>> = OnceLock::new();

127

128fn deferred_slots() -> &'static Arc<Semaphore> {

129 DEFERRED.get_or_init(|| Arc::new(Semaphore::new(MAX_DEFERRED)))

130}

122131

123132/// 启动固定 worker 池；必须在 tokio 运行时内调用一次。

124133pub fn start(service: ChatService, cancel: CancellationToken) {

@@ -194,6 +203,15 @@

194203 match tx.try_send(job) {

195204 Ok(()) => oltp::metrics::chat_fanout(lane.name(), "queued"),

196205 Err(mpsc::error::TrySendError::Full(job)) => {

206 let Ok(permit) = deferred_slots().clone().try_acquire_owned() else {

207 tracing::warn!(

208 kind = job.kind(),

209 lane = lane.name(),

210 "chat fanout deferral saturated; dropping job"

211 );

212 oltp::metrics::chat_fanout(lane.name(), "dropped");

213 return;

214 };

197215 tracing::warn!(

198216 kind = job.kind(),

199217 lane = lane.name(),

@@ -202,6 +220,7 @@

202220 oltp::metrics::chat_fanout(lane.name(), "deferred");

203221 let tx = tx.clone();

204222 tokio::spawn(async move {

223 let _permit = permit;

205224 if tx.send(job).await.is_err() {

206225 tracing::warn!(lane = lane.name(), "chat fanout closed while deferring job");

207226 oltp::metrics::chat_fanout(lane.name(), "closed");

Modified lib/service/deletion/src/lib.rs +33 −13 View lib/service/deletion/src/lib.rs

@@ -236,6 +236,18 @@

236236 user_id: i64,

237237) -> Result<(), DeletionError> {

238238 let db = state.database.write();

239 let existing = account_deletion::Entity::find()

240 .filter(account_deletion::Column::UserId.eq(user_id))

241 .one(db)

242 .await?;

243 if let Some(row) = &existing

244 && row.lease_id.is_some()

245 && row.leased_until.is_some_and(|until| until > now())

246 {

247 return Err(DeletionError::Conflict(

248 "account deletion is currently being purged".into(),

249 ));

250 }

239251 let deleted = account_deletion::Entity::delete_many()

240252 .filter(account_deletion::Column::UserId.eq(user_id))

241253 .filter(

@@ -245,22 +257,30 @@

245257 )

246258 .exec(db)

247259 .await?;

248 if deleted.rows_affected == 0 {

249 let exists = account_deletion::Entity::find()

250 .filter(account_deletion::Column::UserId.eq(user_id))

251 .one(db)

252 .await?

253 .is_some();

254 if exists {

255 return Err(DeletionError::Conflict(

256 "account deletion is currently being purged".into(),

257 ));

258 }

260 if deleted.rows_affected == 0 && existing.is_some() {

261 return Err(DeletionError::Conflict(

262 "account deletion is currently being purged".into(),

263 ));

259264 }

260 accounts

265 // 先删本地行再撤销远端：这样 purge worker 不会再领到它；远端失败时必须把行

266 // 放回去，否则本地记录消失、KenAuth 侧仍会继续删除账号，两侧状态分叉。

267 if let Err(error) = accounts

261268 .cancel_account_deletion(user_id)

262269 .await

263 .map_err(ken_auth)?;

270 .map_err(ken_auth)

271 {

272 if let Some(row) = existing {

273 let restore = account_deletion::ActiveModel::from(row);

274 if let Err(restore_error) = restore.insert(db).await {

275 tracing::error!(

276 %restore_error,

277 user_id,

278 "failed to restore account deletion row after remote cancel failed"

279 );

280 }

281 }

282 return Err(error);

283 }

264284 Ok(())

265285}

266286

Modified lib/service/git/src/handler/commits/list.rs +6 −5 View lib/service/git/src/handler/commits/list.rs

@@ -5,13 +5,13 @@

55use crate::errors::{ApiError, GitApiError};

66use crate::handler;

77use crate::service::GitService;

8use crate::service::commits::{CommitDto, ListCommitsCmd};

8use crate::service::commits::{CommitsPage, ListCommitsCmd};

99

1010#[utoipa::path(

1111 get,

1212 path = "/repos/{space_name}/{repo_name}/commits",

1313 tag = "commits",

14 description = "按修订、路径、时间窗和首父条件分页查询提交历史。",

14 description = "按修订、路径、时间窗和首父条件分页查询提交历史；附带 has_more 与调用方的 diff_view_style。",

1515 params(

1616 crate::handler::RepoPath,

1717 ("revision" = Option<String>, Query, description = "start revision"),

@@ -23,7 +23,7 @@

2323 ("first_parent" = bool, Query, description = "first parent only"),

2424 ),

2525 responses(

26 (status = 200, description = "list commits", body = AppResponseData<api::pagination::Page<CommitDto>>)

26 (status = 200, description = "list commits", body = AppResponseData<CommitsPage>)

2727 )

2828)]

2929pub async fn list_commits(

@@ -31,7 +31,7 @@

3131 OptionalUser(viewer): OptionalUser,

3232 Path(path): Path<crate::handler::RepoPath>,

3333 Query(cmd): Query<ListCommitsCmd>,

34) -> Result<AppResponseData<api::pagination::Page<CommitDto>>, ApiError> {

34) -> Result<AppResponseData<CommitsPage>, ApiError> {

3535 let (ctx, _viewer) = handler::resolve_reader(

3636 &service,

3737 viewer,

@@ -43,5 +43,6 @@

4343 .list_commits(&ctx, cmd)

4444 .await

4545 .map_err(GitApiError::from)?;

46 Ok(AppResponseData::ok(out))

46 let diff_view_style = service.user_diff_view_style(viewer).await;

47 Ok(AppResponseData::ok(CommitsPage::new(out, diff_view_style)))

4748}

Modified lib/service/git/src/service/commits.rs +25 −0 View lib/service/git/src/service/commits.rs

@@ -71,6 +71,31 @@

7171 }

7272}

7373

74/// 提交历史页：统一分页字段 + 该端点特有的 `has_more` / `diff_view_style`。

75#[derive(Debug, Clone, Serialize, utoipa::ToSchema)]

76pub struct CommitsPage {

77 pub items: Vec<CommitDto>,

78 pub total: u64,

79 pub limit: u64,

80 pub offset: u64,

81 pub has_more: bool,

82 pub diff_view_style: String,

83}

84

85impl CommitsPage {

86 pub fn new(page: Page<CommitDto>, diff_view_style: String) -> Self {

87 let has_more = page.offset.saturating_add(page.items.len() as u64) < page.total;

88 Self {

89 items: page.items,

90 total: page.total,

91 limit: page.limit,

92 offset: page.offset,

93 has_more,

94 diff_view_style,

95 }

96 }

97}

98

7499#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)]

75100#[serde(rename_all = "lowercase")]

76101pub enum CommitStatusDtoState {

Modified lib/service/git/src/openapi.rs +0 −1 View lib/service/git/src/openapi.rs

@@ -116,7 +116,6 @@

116116 api::pagination::Page<crate::service::repos::RepoSummaryDto>,

117117 api::pagination::Page<crate::service::branches::BranchDto>,

118118 api::pagination::Page<crate::service::tags::TagDto>,

119 api::pagination::Page<crate::service::commits::CommitDto>,

120119 api::pagination::Page<crate::service::licenses::LicenseDto>,

121120 api::pagination::Page<crate::service::stars::StargazerDto>,

122121 api::pagination::Page<crate::service::watches::WatcherDto>,

Modified lib/share/models/schema_pre_sync.sql +47 −0 View lib/share/models/schema_pre_sync.sql

@@ -30,6 +30,53 @@

3030DROP INDEX IF EXISTS idx_chat_gdm_key;

3131DROP INDEX IF EXISTS idx_chat_draft_unique;

3232DROP INDEX IF EXISTS idx_space_join_requests_pending;

33

34-- 上面被删掉的部分唯一索引在旧版本里可能「删了没建成」（extras 曾整份回滚），

35-- 期间写入的重复行会让 sync 自建唯一键、以及 extras 重建部分索引时失败。

36-- 因此先按业务键收敛到最新一行；表不存在（全新库）时整段跳过。

37DO $$ BEGIN

38 IF to_regclass('issue_watch') IS NOT NULL THEN

39 DELETE FROM issue_watch AS stale

40 USING issue_watch AS keep

41 WHERE stale.issue_id IS NOT NULL

42 AND stale.issue_id = keep.issue_id

43 AND stale.user_id = keep.user_id

44 AND (stale.created_at, stale.id) < (keep.created_at, keep.id);

45

46 DELETE FROM issue_watch AS stale

47 USING issue_watch AS keep

48 WHERE stale.project_id IS NOT NULL

49 AND stale.project_id = keep.project_id

50 AND stale.user_id = keep.user_id

51 AND (stale.created_at, stale.id) < (keep.created_at, keep.id);

52 END IF;

53

54 IF to_regclass('issue_reaction') IS NOT NULL THEN

55 DELETE FROM issue_reaction AS stale

56 USING issue_reaction AS keep

57 WHERE stale.issue_id IS NOT NULL

58 AND stale.issue_id = keep.issue_id

59 AND stale.user_id = keep.user_id

60 AND stale.emoji = keep.emoji

61 AND (stale.created_at, stale.id) < (keep.created_at, keep.id);

62

63 DELETE FROM issue_reaction AS stale

64 USING issue_reaction AS keep

65 WHERE stale.comment_id IS NOT NULL

66 AND stale.comment_id = keep.comment_id

67 AND stale.user_id = keep.user_id

68 AND stale.emoji = keep.emoji

69 AND (stale.created_at, stale.id) < (keep.created_at, keep.id);

70

71 DELETE FROM issue_reaction AS stale

72 USING issue_reaction AS keep

73 WHERE stale.project_id IS NOT NULL

74 AND stale.project_id = keep.project_id

75 AND stale.user_id = keep.user_id

76 AND stale.emoji = keep.emoji

77 AND (stale.created_at, stale.id) < (keep.created_at, keep.id);

78 END IF;

79END $$;

3380-- scheduled messages feature removed; drop the legacy table before sync

3481DROP TABLE IF EXISTS user_scheduled_message;

3582

Modified openapi.json View openapi.json

Diff too large to display.