Refactor Kafka publishing without DB outbox

docs/IM-Kafka接入方案.md

@@ -722,14 +722,14 @@ IM Server 已接入 `im.message.persist`，用于把在线发送链路和最终
 
 当前处理方式：
 
-1. WS/HTTP 发送入口先同步完成权限校验、业务幂等、`seq_id` 分配和 Topic 游标占用。
-2. Kafka 开启时，发送入口先写租户库 `im_kafka_outbox`，由 Outbox worker 可靠生产 `im.message.persist`，客户端仍立即拿到已分配的 `seq_id`。
+1. WS/HTTP 发送入口先完成权限校验，并由 topic 内存态分配 `seq_id`。
+2. Kafka 开启时，发送入口直接生产 `im.message.persist`，不查库、不落库、不写 DB outbox；客户端在 Kafka 写入成功后拿到已分配的 `seq_id`。
 3. Kafka 关闭时，继续走原同步落库路径，兼容本地和未接 Kafka 环境。
 4. 消费者按 `message_kind` 区分私聊与群聊：
    - `p2p` 写入 `im_messages`。
    - `group` 写入 `im_group_messages`。
 5. 消费侧复用 `im_kafka_consume_log` 做 `event_id + consumer_group` 幂等。
-6. 失败按统一退避重试，超过上限写入 `im_kafka_dead_letters` 并投递 `im.dead.letter`。
+6. 失败按统一退避重试 3 次，超过上限先投递 `im.dead.letter`，再由 dead letter consumer 写入 `im_kafka_dead_letters`。
 
 ### 11.10 IM Server 已接入的缓存失效消费
 
@@ -874,17 +874,9 @@ IM Server 已在 `im.admin.command` 中接入用户治理类后台动作。Admin
 }
 ```
 
-### 11.9 IM Server 已接入的 Kafka Outbox Worker
+### 11.9 IM Server 已移除 Kafka DB Outbox
 
-IM Server 已接入租户库 `im_kafka_outbox` 通用发送 worker，用于避免“业务库写成功但 Kafka 未发送”的事件丢失。
-
-当前能力：
-
-1. 启动后按 active 租户扫描每个租户库的 `im_kafka_outbox`。
-2. 扫描 `pending/failed` 且已到 `next_retry_at` 的事件。
-3. 发送成功后标记 `sent`、写入 `sent_at`。
-4. 发送失败后标记 `failed`，按 Kafka 配置的 `retry_backoff_ms` 写入下一次重试时间。
-5. 首个接入方为 `im.message.persist`，后续文件生命周期、工作台事件、审计事件可复用同一 outbox。
+IM Server 不再维护租户库 `im_kafka_outbox`，Kafka 事件统一由发送入口直接生产。生产侧使用 Kafka writer 的 `max_retry_count` 控制发送重试；消费侧使用 `im_kafka_consume_log` 做处理幂等和失败重试，超过上限后进入 `im.dead.letter`。
 
 本轮暂不实现 `im.message.push` 离线 Push Kafka 化，已记录到 [IM 第一至十一章后续增强项](./IM-未实现后续增强项.md)。
 
@@ -930,7 +922,7 @@ IM Server 已接入 `im.workbench.event` 的首批生产与消费。
 
 ### 11.12 IM Server 已接入的审计事件
 
-IM Server 已接入 `im.audit.event` 的可靠生产，审计事件写入租户库 `im_kafka_outbox` 后由 Kafka Outbox worker 发送。
+IM Server 已接入 `im.audit.event` 的直接生产，审计事件不写 DB outbox。
 
 当前审计范围：
 

docs/IM-Kafka接入联调测试文档.md

@@ -54,7 +54,6 @@ SOURCE docs/sql/IM-Kafka基础表SQL.sql;
 | 表 | 用途 |
 | --- | --- |
 | `im_kafka_consume_log` | 消费幂等、重试状态。 |
-| `im_kafka_outbox` | IM 侧可靠生产 Kafka 事件。 |
 | `im_kafka_dead_letters` | 死信记录与人工处理状态。 |
 | `im_messages` | 私聊消息存储。 |
 | `im_group_messages` | 群聊消息存储。 |
@@ -886,15 +885,7 @@ GET /v1/internal/admin-query/tenants/default/kafka/dead-letters/{dead_id}
 
 ### TC-KAFKA-030 Outbox 可靠投递
 
-目标：验证 IM 侧 Kafka Outbox 防止业务成功但 Kafka 丢事件。
-
-步骤：
-
-1. 临时停止 Kafka。
-2. 执行会产生 outbox 的操作，如发送消息、提交工作台事件。
-3. 查询 `im_kafka_outbox`。
-4. 恢复 Kafka。
-5. 等待 outbox worker 投递。
+目标：验证 IM 侧 Kafka 直接生产失败时不会产生 DB outbox，调用方收到发送失败。
 
 预期：
 
@@ -958,7 +949,7 @@ GET /v1/internal/admin-query/tenants/default/kafka/dead-letters/{dead_id}
 | 进入死信 | 查 `im_kafka_dead_letters.envelope/last_error`，修复后用重放命令。 |
 | 用户没有被踢下线 | 查目标 `user_id` 是否是 IM 用户 ID；是否连接在当前 IM 节点；跨节点踢人是否需要后续 Redis fanout 扩展。 |
 | 群治理未生效 | 查 `topic/version/actor_user_id` 是否正确，目标用户是否为群成员。 |
-| 消息未落库 | 查 `im_kafka_outbox` 是否 sent，`im.message.persist` 是否消费成功。 |
+| 消息未落库 | 查 `im.message.persist` 是否消费成功，以及 `im_kafka_consume_log.last_reason/last_error`。 |
 | 附件未绑定 | 查消息是否带附件、`im.file.lifecycle` 是否消费成功。 |
 | 内部 API 401 | 查 secret、签名文本、RequestURI、timestamp、nonce 是否一致。 |
 | 配置没有刷新 | 查是否发送 `im.config.event` 或 `im.cache.invalidate`，IM 是否清理了对应 namespace。 |
@@ -1023,7 +1014,7 @@ GET /v1/internal/admin-query/tenants/default/kafka/dead-letters/{dead_id}
 | 文件生命周期 | 上传完成、附件绑定、群文件登记/删除、OSS 元数据和引用表。 |
 | 工作台事件 | 会议、日报/月报、总经理信箱、红点/系统通知投递。 |
 | 死信闭环 | 死信查询、忽略、解决、重放，以及重放后二次成功/失败。 |
-| Outbox | Kafka 停止/恢复时 `im_kafka_outbox`、Admin outbox 的可靠投递。 |
+| Kafka 生产失败 | Kafka 停止/恢复时，确认 IM 不写 `im_kafka_outbox`，接口/WS 返回发送失败，恢复后可重新发送。 |
 | Admin 页面 | 操作单状态从 pending/running 到 succeeded/failed/dead 的页面联调。 |
 
 ### 10.2 2026-08-27 完整后端链路联调

docs/sql/IM-Kafka-Outbox删除SQL.sql

@@ -0,0 +1,4 @@
+-- Remove the deprecated Kafka DB outbox table from a tenant database.
+-- Run once per tenant business database after deploying the direct Kafka publish path.
+
+DROP TABLE IF EXISTS im_kafka_outbox;

docs/sql/IM-Kafka基础表SQL.sql

@@ -29,29 +29,6 @@ CREATE TABLE IF NOT EXISTS im_kafka_consume_log (
   INDEX idx_im_kafka_consume_log_topic(source_topic, id)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
 
-CREATE TABLE IF NOT EXISTS im_kafka_outbox (
-  id BIGINT NOT NULL AUTO_INCREMENT,
-  event_id VARCHAR(96) NOT NULL,
-  topic VARCHAR(128) NOT NULL,
-  event_type VARCHAR(96) NOT NULL,
-  biz_type VARCHAR(64) NOT NULL DEFAULT '',
-  biz_id VARCHAR(128) NOT NULL DEFAULT '',
-  key_value VARCHAR(256) NOT NULL DEFAULT '',
-  envelope JSON NOT NULL,
-  status VARCHAR(16) NOT NULL DEFAULT 'pending',
-  attempts INT NOT NULL DEFAULT 0,
-  next_retry_at DATETIME(3),
-  sent_at DATETIME(3),
-  last_reason VARCHAR(64) NOT NULL DEFAULT '',
-  last_error VARCHAR(1000) NOT NULL DEFAULT '',
-  created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
-  updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
-  PRIMARY KEY(id),
-  UNIQUE INDEX uk_im_kafka_outbox_event(event_id),
-  INDEX idx_im_kafka_outbox_pending(status, next_retry_at, id),
-  INDEX idx_im_kafka_outbox_biz(biz_type, biz_id)
-) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
-
 CREATE TABLE IF NOT EXISTS im_kafka_dead_letters (
   id BIGINT NOT NULL AUTO_INCREMENT,
   dead_id VARCHAR(96) NOT NULL,

server/db/adapter.go

@@ -263,10 +263,6 @@ type TenantHTTPInfrastructureAdapter interface {
 	TenantKafkaDeadLetterGet(t.TenantID, string) (*t.KafkaDeadLetter, error)
 	TenantKafkaDeadLetterMark(t.TenantID, string, string, string, string, time.Time) error
 	TenantKafkaDeadLetterMarkReplay(t.TenantID, string, string, string, time.Time) (*t.KafkaDeadLetter, error)
-	TenantKafkaOutboxCreate(t.TenantID, *t.KafkaOutboxRecord) error
-	TenantKafkaOutboxListPending(t.TenantID, int, time.Time) ([]t.KafkaOutboxRecord, error)
-	TenantKafkaOutboxMarkSent(t.TenantID, string, time.Time) error
-	TenantKafkaOutboxMarkFailed(t.TenantID, string, string, string, time.Time, time.Time) error
 }
 
 // TenantRegistrationInviteAdapter stores invitation definitions in the

server/db/mysql/infrastructure_schema.go

@@ -44,20 +44,6 @@ func ensureIMHTTPInfrastructureTables(exec sqlExecutor) error {
 			INDEX idx_im_kafka_consume_log_biz(biz_type, biz_id),
 			INDEX idx_im_kafka_consume_log_topic(source_topic, id)
 		) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`,
-		`CREATE TABLE IF NOT EXISTS im_kafka_outbox (
-			id BIGINT NOT NULL AUTO_INCREMENT, event_id VARCHAR(96) NOT NULL,
-			topic VARCHAR(128) NOT NULL, event_type VARCHAR(96) NOT NULL,
-			biz_type VARCHAR(64) NOT NULL DEFAULT '', biz_id VARCHAR(128) NOT NULL DEFAULT '',
-			key_value VARCHAR(256) NOT NULL DEFAULT '', envelope JSON NOT NULL,
-			status VARCHAR(16) NOT NULL DEFAULT 'pending', attempts INT NOT NULL DEFAULT 0,
-			next_retry_at DATETIME(3), sent_at DATETIME(3), last_reason VARCHAR(64) NOT NULL DEFAULT '',
-			last_error VARCHAR(1000) NOT NULL DEFAULT '',
-			created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
-			updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
-			PRIMARY KEY(id), UNIQUE INDEX uk_im_kafka_outbox_event(event_id),
-			INDEX idx_im_kafka_outbox_pending(status, next_retry_at, id),
-			INDEX idx_im_kafka_outbox_biz(biz_type, biz_id)
-		) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`,
 		`CREATE TABLE IF NOT EXISTS im_kafka_dead_letters (
 			id BIGINT NOT NULL AUTO_INCREMENT, dead_id VARCHAR(96) NOT NULL,
 			event_id VARCHAR(96) NOT NULL, parent_event_id VARCHAR(96) NOT NULL DEFAULT '',

server/db/mysql/infrastructure_schema_test.go

@@ -12,15 +12,14 @@ func TestEnsureIMHTTPInfrastructureTablesUsesSelectedTenantDatabase(t *testing.T
 	if err := ensureIMHTTPInfrastructureTables(exec); err != nil {
 		t.Fatal(err)
 	}
-	if len(exec.statements) != 8 {
-		t.Fatalf("created %d infrastructure tables, want 8", len(exec.statements))
+	if len(exec.statements) != 7 {
+		t.Fatalf("created %d infrastructure tables, want 7", len(exec.statements))
 	}
 	joined := strings.ToLower(strings.Join(exec.statements, "\n"))
 	for _, table := range []string{
 		"im_http_idempotency",
 		"im_business_events_outbox",
 		"im_kafka_consume_log",
-		"im_kafka_outbox",
 		"im_kafka_dead_letters",
 		"im_realname_appeals",
 		"im_conversation_settings",
@@ -33,4 +32,7 @@ func TestEnsureIMHTTPInfrastructureTablesUsesSelectedTenantDatabase(t *testing.T
 	if strings.Contains(joined, "tinode.") || strings.Contains(joined, "create database") || strings.Contains(joined, "tenant_id") {
 		t.Fatalf("infrastructure schema must stay inside the selected tenant database: %s", joined)
 	}
+	if strings.Contains(joined, "im_kafka_outbox") {
+		t.Fatalf("kafka outbox table must not be created: %s", joined)
+	}
 }

server/db/mysql/tenant_infrastructure.go

@@ -512,140 +512,6 @@ func (a *adapter) TenantKafkaDeadLetterMarkReplay(id t.TenantID, deadID, resolve
 	return a.TenantKafkaDeadLetterGet(id, deadID)
 }
 
-func (a *adapter) TenantKafkaOutboxCreate(id t.TenantID, record *t.KafkaOutboxRecord) error {
-	tenantAdp, err := a.infrastructureAdapter(id)
-	if err != nil {
-		return err
-	}
-	if record == nil || record.EventID == "" || record.Topic == "" || record.EventType == "" || len(record.Envelope) == 0 {
-		return t.ErrMalformed
-	}
-	if tenantAdp != a {
-		return tenantAdp.TenantKafkaOutboxCreate(id, record)
-	}
-	now := t.TimeNow()
-	_, err = a.db.Exec(`INSERT INTO im_kafka_outbox
-		(event_id,topic,event_type,biz_type,biz_id,key_value,envelope,status,next_retry_at,created_at,updated_at)
-		VALUES(?,?,?,?,?,?,?,?,?,?,?)
-		ON DUPLICATE KEY UPDATE updated_at=updated_at`,
-		record.EventID, record.Topic, record.EventType, record.BizType, record.BizID, record.Key,
-		record.Envelope, t.KafkaOutboxPending, record.NextRetryAt, now, now)
-	return err
-}
-
-func (a *adapter) TenantKafkaOutboxListPending(id t.TenantID, limit int, now time.Time) ([]t.KafkaOutboxRecord, error) {
-	tenantAdp, err := a.infrastructureAdapter(id)
-	if err != nil {
-		return nil, err
-	}
-	if limit <= 0 {
-		limit = 100
-	}
-	if tenantAdp != a {
-		return tenantAdp.TenantKafkaOutboxListPending(id, limit, now)
-	}
-	rows := []struct {
-		ID          int64        `db:"id"`
-		EventID     string       `db:"event_id"`
-		Topic       string       `db:"topic"`
-		EventType   string       `db:"event_type"`
-		BizType     string       `db:"biz_type"`
-		BizID       string       `db:"biz_id"`
-		Key         string       `db:"key_value"`
-		Envelope    []byte       `db:"envelope"`
-		Status      string       `db:"status"`
-		Attempts    int          `db:"attempts"`
-		NextRetryAt sql.NullTime `db:"next_retry_at"`
-		SentAt      sql.NullTime `db:"sent_at"`
-		LastReason  string       `db:"last_reason"`
-		LastError   string       `db:"last_error"`
-		CreatedAt   time.Time    `db:"created_at"`
-		UpdatedAt   time.Time    `db:"updated_at"`
-	}{}
-	err = a.db.Select(&rows, `SELECT id,event_id,topic,event_type,biz_type,biz_id,key_value,envelope,status,attempts,
-		next_retry_at,sent_at,last_reason,last_error,created_at,updated_at
-		FROM im_kafka_outbox
-		WHERE status IN (?,?) AND (next_retry_at IS NULL OR next_retry_at<=?)
-		ORDER BY id ASC LIMIT ?`, t.KafkaOutboxPending, t.KafkaOutboxFailed, now, limit)
-	if err != nil {
-		return nil, err
-	}
-	out := make([]t.KafkaOutboxRecord, 0, len(rows))
-	for _, row := range rows {
-		rec := t.KafkaOutboxRecord{
-			ID:         row.ID,
-			EventID:    row.EventID,
-			Topic:      row.Topic,
-			EventType:  row.EventType,
-			BizType:    row.BizType,
-			BizID:      row.BizID,
-			Key:        row.Key,
-			Envelope:   row.Envelope,
-			Status:     row.Status,
-			Attempts:   row.Attempts,
-			LastReason: row.LastReason,
-			LastError:  row.LastError,
-			CreatedAt:  row.CreatedAt,
-			UpdatedAt:  row.UpdatedAt,
-		}
-		if row.NextRetryAt.Valid {
-			rec.NextRetryAt = &row.NextRetryAt.Time
-		}
-		if row.SentAt.Valid {
-			rec.SentAt = &row.SentAt.Time
-		}
-		out = append(out, rec)
-	}
-	return out, nil
-}
-
-func (a *adapter) TenantKafkaOutboxMarkSent(id t.TenantID, eventID string, now time.Time) error {
-	tenantAdp, err := a.infrastructureAdapter(id)
-	if err != nil {
-		return err
-	}
-	if eventID == "" {
-		return t.ErrMalformed
-	}
-	if tenantAdp != a {
-		return tenantAdp.TenantKafkaOutboxMarkSent(id, eventID, now)
-	}
-	res, err := a.db.Exec(`UPDATE im_kafka_outbox SET status=?,sent_at=?,attempts=attempts+1,last_reason='',
-		last_error='',updated_at=? WHERE event_id=? AND status<>?`,
-		t.KafkaOutboxSent, now, now, eventID, t.KafkaOutboxSent)
-	if err != nil {
-		return err
-	}
-	if affected, _ := res.RowsAffected(); affected == 0 {
-		return t.ErrNotFound
-	}
-	return nil
-}
-
-func (a *adapter) TenantKafkaOutboxMarkFailed(id t.TenantID, eventID, reason, message string, nextRetryAt, now time.Time) error {
-	tenantAdp, err := a.infrastructureAdapter(id)
-	if err != nil {
-		return err
-	}
-	if eventID == "" {
-		return t.ErrMalformed
-	}
-	if tenantAdp != a {
-		return tenantAdp.TenantKafkaOutboxMarkFailed(id, eventID, reason, message, nextRetryAt, now)
-	}
-	res, err := a.db.Exec(`UPDATE im_kafka_outbox SET status=?,attempts=attempts+1,next_retry_at=?,
-		last_reason=?,last_error=?,updated_at=? WHERE event_id=? AND status<>?`,
-		t.KafkaOutboxFailed, nextRetryAt, truncateKafkaString(reason, 64), truncateKafkaString(message, 1000),
-		now, eventID, t.KafkaOutboxSent)
-	if err != nil {
-		return err
-	}
-	if affected, _ := res.RowsAffected(); affected == 0 {
-		return t.ErrNotFound
-	}
-	return nil
-}
-
 func scanKafkaDeadLetter(rows *sqlx.Rows) (*t.KafkaDeadLetter, error) {
 	var row struct {
 		ID            int64     `db:"id"`

server/im_http_test.go

@@ -38,9 +38,9 @@ type testVerificationStore struct {
 }
 
 type testHTTPInfrastructureStore struct {
-	records map[string]*types.HTTPIdempotencyRecord
-	events  []types.BusinessEvent
-	outbox  []types.KafkaOutboxRecord
+	records     map[string]*types.HTTPIdempotencyRecord
+	events      []types.BusinessEvent
+	deadLetters []types.KafkaDeadLetter
 }
 
 type testTenantEntryStore struct {
@@ -162,7 +162,10 @@ func (s *testHTTPInfrastructureStore) MarkKafkaConsumeFailed(_ types.TenantID, e
 	return &types.KafkaConsumeRecord{EventID: eventID, ConsumerGroup: group, Status: types.KafkaConsumeFailed, RetryCount: 1, LastReason: reason, LastError: message}, nil
 }
 
-func (s *testHTTPInfrastructureStore) MarkKafkaConsumeDead(_ types.TenantID, _ *types.KafkaDeadLetter, _ time.Time) error {
+func (s *testHTTPInfrastructureStore) MarkKafkaConsumeDead(_ types.TenantID, dead *types.KafkaDeadLetter, _ time.Time) error {
+	if dead != nil {
+		s.deadLetters = append(s.deadLetters, *dead)
+	}
 	return nil
 }
 
@@ -182,23 +185,6 @@ func (s *testHTTPInfrastructureStore) MarkKafkaDeadLetterReplay(_ types.TenantID
 	return &types.KafkaDeadLetter{DeadID: deadID, Status: types.KafkaDeadStatusPendingReplay}, nil
 }
 
-func (s *testHTTPInfrastructureStore) CreateKafkaOutbox(_ types.TenantID, record *types.KafkaOutboxRecord) error {
-	s.outbox = append(s.outbox, *record)
-	return nil
-}
-
-func (s *testHTTPInfrastructureStore) ListPendingKafkaOutbox(_ types.TenantID, _ int, _ time.Time) ([]types.KafkaOutboxRecord, error) {
-	return append([]types.KafkaOutboxRecord(nil), s.outbox...), nil
-}
-
-func (s *testHTTPInfrastructureStore) MarkKafkaOutboxSent(_ types.TenantID, _ string, _ time.Time) error {
-	return nil
-}
-
-func (s *testHTTPInfrastructureStore) MarkKafkaOutboxFailed(_ types.TenantID, _, _, _ string, _, _ time.Time) error {
-	return nil
-}
-
 type testRelationsStore struct {
 	context        *types.RelationshipContext
 	created        *types.FriendRequest

server/kafka_admin_commands.go

@@ -38,6 +38,8 @@ const (
 	adminCommandKafkaDeadResolve   = "kafka.dead_letter.resolve"
 )
 
+var publishKafkaDeadLetter = publishKafkaEnvelope
+
 type adminCommandPayload struct {
 	CommandID    string               `json:"command_id"`
 	ActorUserID  string               `json:"actor_user_id"`
@@ -482,6 +484,13 @@ func kafkaTenantID(tenantCode string) (types.TenantID, error) {
 	return tenant.ID, nil
 }
 
+func kafkaTenantIDFromEnvelope(envelope kafkaEnvelope) (types.TenantID, error) {
+	if !envelope.TenantID.IsZero() {
+		return envelope.TenantID, nil
+	}
+	return kafkaTenantID(envelope.TenantCode)
+}
+
 func markKafkaDeadLetter(ctx context.Context, tenantID types.TenantID, envelope kafkaEnvelope, consume *types.KafkaConsumeRecord, reason, message string) error {
 	encoded, err := json.Marshal(envelope)
 	if err != nil {
@@ -512,26 +521,15 @@ func markKafkaDeadLetter(ctx context.Context, tenantID types.TenantID, envelope
 		Status:        types.KafkaDeadStatusDead,
 	}
 	now := time.Now().UTC().Round(time.Millisecond)
-	if err = store.HTTPInfrastructure.MarkKafkaConsumeDead(tenantID, dead, now); err != nil {
-		return err
-	}
-	if auditErr := enqueueKafkaAuditEvent(ctx, tenantID, "system", types.ZeroUid, "kafka.dead_letter.created", "kafka_event", envelope.EventID, "dead", reason, map[string]any{
-		"dead_id":        deadID,
-		"event_type":     envelope.EventType,
-		"biz_type":       envelope.BizType,
-		"biz_id":         envelope.BizID,
-		"consumer_group": kafkaBus.config.ConsumerGroup,
-	}); auditErr != nil {
-		logs.Warn.Printf("kafka dead letter audit enqueue failed: event=%s dead=%s err=%v", envelope.EventID, deadID, auditErr)
-	}
 	deadPayload, err := json.Marshal(dead)
 	if err != nil {
 		return err
 	}
-	return publishKafkaEnvelope(ctx, "dead_letter", envelope.EventID, kafkaEnvelope{
+	return publishKafkaDeadLetter(ctx, "dead_letter", envelope.EventID, kafkaEnvelope{
 		EventID:       deadID,
 		TraceID:       envelope.TraceID,
 		TenantCode:    envelope.TenantCode,
+		TenantID:      tenantID,
 		SourceService: "im-server",
 		TargetService: "hotgo-admin",
 		EventType:     "kafka.dead_letter.created",

server/kafka_dead_letters.go

@@ -6,10 +6,57 @@ import (
 	"strings"
 	"time"
 
+	"github.com/segmentio/kafka-go"
+	"github.com/tinode/chat/server/logs"
 	"github.com/tinode/chat/server/store"
 	"github.com/tinode/chat/server/store/types"
 )
 
+func startKafkaDeadLetterConsumer() {
+	startKafkaConsumer("dead_letter", handleKafkaDeadLetter)
+}
+
+func handleKafkaDeadLetter(ctx context.Context, envelope kafkaEnvelope, _ kafka.Message) error {
+	var dead types.KafkaDeadLetter
+	if err := json.Unmarshal(envelope.Payload, &dead); err != nil {
+		return err
+	}
+	if dead.DeadID == "" {
+		dead.DeadID = envelope.EventID
+	}
+	tenantID, err := tenantIDFromDeadLetter(envelope, &dead)
+	if err != nil {
+		return err
+	}
+	now := time.Now().UTC().Round(time.Millisecond)
+	if err = store.HTTPInfrastructure.MarkKafkaConsumeDead(tenantID, &dead, now); err != nil {
+		return err
+	}
+	if auditErr := enqueueKafkaAuditEvent(ctx, tenantID, "system", types.ZeroUid, "kafka.dead_letter.created", "kafka_event", dead.EventID, "dead", dead.LastReason, map[string]any{
+		"dead_id":        dead.DeadID,
+		"event_type":     dead.EventType,
+		"biz_type":       dead.BizType,
+		"biz_id":         dead.BizID,
+		"consumer_group": dead.ConsumerGroup,
+	}); auditErr != nil {
+		logs.Warn.Printf("kafka dead letter audit enqueue failed: event=%s dead=%s err=%v", dead.EventID, dead.DeadID, auditErr)
+	}
+	return nil
+}
+
+func tenantIDFromDeadLetter(envelope kafkaEnvelope, dead *types.KafkaDeadLetter) (types.TenantID, error) {
+	if !envelope.TenantID.IsZero() {
+		return envelope.TenantID, nil
+	}
+	if dead != nil && len(dead.Envelope) > 0 {
+		var original kafkaEnvelope
+		if err := json.Unmarshal(dead.Envelope, &original); err == nil && !original.TenantID.IsZero() {
+			return original.TenantID, nil
+		}
+	}
+	return kafkaTenantID(envelope.TenantCode)
+}
+
 func replayKafkaDeadLetter(ctx context.Context, tenantID types.TenantID, deadID string, actor types.Uid, note string) (*types.KafkaDeadLetter, error) {
 	now := time.Now().UTC().Round(time.Millisecond)
 	dead, err := store.HTTPInfrastructure.MarkKafkaDeadLetterReplay(tenantID, deadID, actor.UserId(), note, now)

server/kafka_file_lifecycle.go

@@ -104,15 +104,7 @@ func enqueueKafkaGroupFileLifecycle(ctx context.Context, tenantID types.TenantID
 	return enqueueKafkaFileLifecycle(ctx, tenantID, eventType, firstNonEmpty(file.GroupFileID, file.FileID), payload)
 }
 
-func enqueueKafkaFileLifecycle(_ context.Context, tenantID types.TenantID, eventType, bizID string, payload kafkaFileLifecyclePayload) error {
-	tenantCode := ""
-	if store.Tenants != nil {
-		if tenant, err := store.Tenants.GetByID(tenantID); err == nil && tenant != nil {
-			tenantCode = tenant.Code
-		} else {
-			logs.Warn.Printf("kafka file lifecycle: failed to resolve tenant code for tenant %d: %v", tenantID, err)
-		}
-	}
+func enqueueKafkaFileLifecycle(ctx context.Context, tenantID types.TenantID, eventType, bizID string, payload kafkaFileLifecyclePayload) error {
 	if payload.OccurredAt.IsZero() {
 		payload.OccurredAt = time.Now().UTC().Round(time.Millisecond)
 	}
@@ -121,9 +113,9 @@ func enqueueKafkaFileLifecycle(_ context.Context, tenantID types.TenantID, event
 		return err
 	}
 	eventID := stableIMEventID("file-lifecycle", tenantID, eventType, bizID, payload.SeqID, strings.Join(payload.AttachmentURLs, ","))
-	err = enqueueKafkaOutbox(tenantID, "file_lifecycle", bizID, kafkaEnvelope{
+	err = publishKafkaEnvelope(ctx, "file_lifecycle", bizID, kafkaEnvelope{
 		EventID:       eventID,
-		TenantCode:    tenantCode,
+		TenantID:      tenantID,
 		SourceService: "im-server",
 		TargetService: "im-server",
 		EventType:     eventType,
@@ -149,7 +141,7 @@ func enqueueKafkaFileLifecycle(_ context.Context, tenantID types.TenantID, event
 }
 
 func handleKafkaFileLifecycle(ctx context.Context, envelope kafkaEnvelope, _ kafka.Message) error {
-	tenantID, err := kafkaTenantID(envelope.TenantCode)
+	tenantID, err := kafkaTenantIDFromEnvelope(envelope)
 	if err != nil {
 		return err
 	}

server/kafka_message_persist.go

@@ -20,6 +20,8 @@ type kafkaMessagePersistPayload struct {
 	MessageKind    string        `json:"message_kind"`
 }
 
+var publishKafkaMessagePersist = publishKafkaEnvelope
+
 func startKafkaMessagePersistConsumer() {
 	startKafkaConsumer("message_persist", handleKafkaMessagePersist)
 }
@@ -28,28 +30,27 @@ func saveMessageMaybeKafka(ctx context.Context, tenantID types.TenantID, msg *ty
 	if !kafkaEnabled() {
 		return store.Messages.Save(tenantID, msg, attachmentURLs, readBySender)
 	}
-	saveResult, err := store.PrepareMessagePersist(tenantID, msg, attachmentURLs, readBySender)
-	if err != nil || saveResult == nil || saveResult.Duplicate {
-		return saveResult, err
+	if msg == nil || msg.TenantID != tenantID || msg.Topic == "" || msg.SeqId <= 0 {
+		return nil, types.ErrMalformed
 	}
-	if readBySender && !types.ParseUid(msg.From).IsZero() {
-		saveResult.MarkedReadBySender = true
+	clientMsgID, err := store.NormalizeClientMsgID(msg.Head)
+	if err != nil {
+		return nil, err
 	}
+	msg.ClientMsgID = clientMsgID
+	msg.InitTimes()
 	if err = enqueueKafkaMessagePersist(ctx, tenantID, msg, attachmentURLs, readBySender); err != nil {
 		return nil, err
 	}
-	return saveResult, nil
-}
-
-func enqueueKafkaMessagePersist(_ context.Context, tenantID types.TenantID, msg *types.Message, attachmentURLs []string, readBySender bool) error {
-	tenantCode := ""
-	if store.Tenants != nil {
-		if tenant, err := store.Tenants.GetByID(tenantID); err == nil && tenant != nil {
-			tenantCode = tenant.Code
-		} else {
-			logs.Warn.Printf("kafka message persist: failed to resolve tenant code for tenant %d: %v", tenantID, err)
+	if types.GetTopicCat(msg.Topic) == types.TopicCatGrp {
+		if cacheErr := store.CachePendingHotMessage(tenantID, msg); cacheErr != nil {
+			logs.Warn.Printf("kafka message persist: failed to cache pending hot message after publish topic=%s seq=%d: %v", msg.Topic, msg.SeqId, cacheErr)
 		}
 	}
+	return &types.MessageSaveResult{SeqId: msg.SeqId, MarkedReadBySender: readBySender && !types.ParseUid(msg.From).IsZero()}, nil
+}
+
+func enqueueKafkaMessagePersist(ctx context.Context, tenantID types.TenantID, msg *types.Message, attachmentURLs []string, readBySender bool) error {
 	kind := "p2p"
 	if types.GetTopicCat(msg.Topic) == types.TopicCatGrp {
 		kind = "group"
@@ -66,7 +67,7 @@ func enqueueKafkaMessagePersist(_ context.Context, tenantID types.TenantID, msg
 	eventID := fmt.Sprintf("msg.persist.%d.%s.%d", tenantID, msg.Topic, msg.SeqId)
 	envelope := kafkaEnvelope{
 		EventID:        eventID,
-		TenantCode:     tenantCode,
+		TenantID:       tenantID,
 		SourceService:  "im-server",
 		TargetService:  "im-server",
 		EventType:      "message.persist." + kind,
@@ -78,13 +79,17 @@ func enqueueKafkaMessagePersist(_ context.Context, tenantID types.TenantID, msg
 		Actor:          &kafkaActor{Type: "user", ID: types.ParseUid(msg.From).UserId()},
 		Payload:        payload,
 	}
-	return enqueueKafkaOutbox(tenantID, "message_persist", msg.Topic, envelope)
+	return publishKafkaMessagePersist(ctx, "message_persist", msg.Topic, envelope)
 }
 
 func handleKafkaMessagePersist(ctx context.Context, envelope kafkaEnvelope, _ kafka.Message) error {
-	tenantID, err := kafkaTenantID(envelope.TenantCode)
-	if err != nil {
-		return err
+	tenantID := envelope.TenantID
+	var err error
+	if tenantID.IsZero() {
+		tenantID, err = kafkaTenantID(envelope.TenantCode)
+		if err != nil {
+			return err
+		}
 	}
 	consume, claimed, err := beginKafkaConsumeForTopic(tenantID, envelope, "message_persist")
 	if err != nil || !claimed {

server/kafka_outbox.go

@@ -1,126 +0,0 @@
-package main
-
-import (
-	"context"
-	"encoding/json"
-	"time"
-
-	"github.com/segmentio/kafka-go"
-	"github.com/tinode/chat/server/logs"
-	"github.com/tinode/chat/server/store"
-	"github.com/tinode/chat/server/store/types"
-)
-
-const (
-	kafkaOutboxPollInterval = 2 * time.Second
-	kafkaOutboxTenantLimit  = 1000
-	kafkaOutboxBatchLimit   = 100
-)
-
-func startKafkaOutboxPoller() {
-	if !kafkaEnabled() {
-		return
-	}
-	kafkaBus.wg.Add(1)
-	go func() {
-		defer kafkaBus.wg.Done()
-		ticker := time.NewTicker(kafkaOutboxPollInterval)
-		defer ticker.Stop()
-		for {
-			select {
-			case <-kafkaBus.ctx.Done():
-				return
-			default:
-				drainKafkaOutbox(kafkaBus.ctx)
-			}
-			select {
-			case <-kafkaBus.ctx.Done():
-				return
-			case <-ticker.C:
-			}
-		}
-	}()
-}
-
-func enqueueKafkaOutbox(tenantID types.TenantID, topicName, key string, envelope kafkaEnvelope) error {
-	if envelope.OccurredAt.IsZero() {
-		envelope.OccurredAt = time.Now().UTC()
-	}
-	raw, err := json.Marshal(envelope)
-	if err != nil {
-		return err
-	}
-	return store.HTTPInfrastructure.CreateKafkaOutbox(tenantID, &types.KafkaOutboxRecord{
-		EventID:     envelope.EventID,
-		Topic:       topicName,
-		EventType:   envelope.EventType,
-		BizType:     envelope.BizType,
-		BizID:       envelope.BizID,
-		Key:         key,
-		Envelope:    raw,
-		Status:      types.KafkaOutboxPending,
-		NextRetryAt: nil,
-	})
-}
-
-func drainKafkaOutbox(ctx context.Context) {
-	if store.Tenants == nil || store.HTTPInfrastructure == nil {
-		return
-	}
-	tenants, err := store.Tenants.ListActive(kafkaOutboxTenantLimit)
-	if err != nil {
-		logs.Warn.Printf("kafka outbox: list active tenants: %v", err)
-		return
-	}
-	now := time.Now().UTC().Round(time.Millisecond)
-	for _, tenant := range tenants {
-		if tenant.ID.IsZero() {
-			continue
-		}
-		records, err := store.HTTPInfrastructure.ListPendingKafkaOutbox(tenant.ID, kafkaOutboxBatchLimit, now)
-		if err != nil {
-			logs.Warn.Printf("kafka outbox: list tenant=%d: %v", tenant.ID, err)
-			continue
-		}
-		for _, record := range records {
-			if err = publishKafkaOutboxRecord(ctx, record); err != nil {
-				nextRetryAt := now.Add(kafkaOutboxRetryDelay(record.Attempts + 1))
-				if markErr := store.HTTPInfrastructure.MarkKafkaOutboxFailed(tenant.ID, record.EventID,
-					"KAFKA_PUBLISH_FAILED", err.Error(), nextRetryAt, now); markErr != nil {
-					logs.Warn.Printf("kafka outbox: mark failed event=%s tenant=%d: %v", record.EventID, tenant.ID, markErr)
-				}
-				continue
-			}
-			if err = store.HTTPInfrastructure.MarkKafkaOutboxSent(tenant.ID, record.EventID, time.Now().UTC().Round(time.Millisecond)); err != nil {
-				logs.Warn.Printf("kafka outbox: mark sent event=%s tenant=%d: %v", record.EventID, tenant.ID, err)
-			}
-		}
-	}
-}
-
-func publishKafkaOutboxRecord(ctx context.Context, record types.KafkaOutboxRecord) error {
-	topic := kafkaTopic(record.Topic)
-	if topic == "" {
-		topic = record.Topic
-	}
-	return kafkaBus.writer.WriteMessages(ctx, kafka.Message{
-		Topic: topic,
-		Key:   []byte(record.Key),
-		Value: record.Envelope,
-		Time:  time.Now().UTC(),
-	})
-}
-
-func kafkaOutboxRetryDelay(attempt int) time.Duration {
-	if attempt <= 0 {
-		attempt = 1
-	}
-	if kafkaBus == nil || len(kafkaBus.config.RetryBackoffMS) == 0 {
-		return time.Duration(attempt) * 10 * time.Second
-	}
-	index := attempt - 1
-	if index >= len(kafkaBus.config.RetryBackoffMS) {
-		index = len(kafkaBus.config.RetryBackoffMS) - 1
-	}
-	return time.Duration(kafkaBus.config.RetryBackoffMS[index]) * time.Millisecond
-}

server/kafka_runtime.go

@@ -13,6 +13,7 @@ import (
 	"github.com/segmentio/kafka-go/sasl"
 	"github.com/segmentio/kafka-go/sasl/scram"
 	"github.com/tinode/chat/server/logs"
+	"github.com/tinode/chat/server/store/types"
 )
 
 const (
@@ -55,6 +56,7 @@ type kafkaEnvelope struct {
 	EventID        string          `json:"event_id"`
 	TraceID        string          `json:"trace_id,omitempty"`
 	TenantCode     string          `json:"tenant_code,omitempty"`
+	TenantID       types.TenantID  `json:"tenant_id,omitempty"`
 	SourceService  string          `json:"source_service,omitempty"`
 	TargetService  string          `json:"target_service,omitempty"`
 	EventType      string          `json:"event_type"`
@@ -132,6 +134,7 @@ func initKafkaRuntime(raw json.RawMessage) error {
 		AllowAutoTopicCreation: cfg.AllowAutoTopic,
 		WriteTimeout:           time.Duration(cfg.WriteTimeoutMS) * time.Millisecond,
 		ReadTimeout:            time.Duration(cfg.ReadTimeoutMS) * time.Millisecond,
+		MaxAttempts:            cfg.MaxRetryCount,
 	}
 
 	if err := rt.ping(context.Background()); err != nil {
@@ -164,7 +167,7 @@ func normalizeKafkaConfig(cfg *kafkaConfig) {
 		cfg.RequiredAcks = int(kafka.RequireOne)
 	}
 	if cfg.MaxRetryCount <= 0 {
-		cfg.MaxRetryCount = 5
+		cfg.MaxRetryCount = 3
 	}
 	if len(cfg.RetryBackoffMS) == 0 {
 		cfg.RetryBackoffMS = []int{10000, 30000, 60000, 300000, 900000}

server/kafka_workbench_audit.go

@@ -6,7 +6,6 @@ import (
 	"time"
 
 	"github.com/segmentio/kafka-go"
-	"github.com/tinode/chat/server/logs"
 	"github.com/tinode/chat/server/store"
 	"github.com/tinode/chat/server/store/types"
 )
@@ -87,15 +86,7 @@ func enqueueKafkaAuditEvent(ctx context.Context, tenantID types.TenantID, actorT
 	return enqueueKafkaTypedEvent(ctx, tenantID, "audit_event", "audit."+action, "audit", resource+":"+resourceID, actor, payload)
 }
 
-func enqueueKafkaTypedEvent(_ context.Context, tenantID types.TenantID, topicName, eventType, bizType, bizID string, actor types.Uid, payload any) error {
-	tenantCode := ""
-	if store.Tenants != nil {
-		if tenant, err := store.Tenants.GetByID(tenantID); err == nil && tenant != nil {
-			tenantCode = tenant.Code
-		} else {
-			logs.Warn.Printf("kafka event: failed to resolve tenant code for tenant %d: %v", tenantID, err)
-		}
-	}
+func enqueueKafkaTypedEvent(ctx context.Context, tenantID types.TenantID, topicName, eventType, bizType, bizID string, actor types.Uid, payload any) error {
 	encoded, err := json.Marshal(payload)
 	if err != nil {
 		return err
@@ -103,7 +94,7 @@ func enqueueKafkaTypedEvent(_ context.Context, tenantID types.TenantID, topicNam
 	eventID := stableIMEventID(topicName, tenantID, eventType, bizID, encoded)
 	envelope := kafkaEnvelope{
 		EventID:       eventID,
-		TenantCode:    tenantCode,
+		TenantID:      tenantID,
 		SourceService: "im-server",
 		TargetService: "im-server",
 		EventType:     eventType,
@@ -115,11 +106,11 @@ func enqueueKafkaTypedEvent(_ context.Context, tenantID types.TenantID, topicNam
 	if !actor.IsZero() {
 		envelope.Actor = &kafkaActor{Type: "user", ID: actor.UserId()}
 	}
-	return enqueueKafkaOutbox(tenantID, topicName, bizID, envelope)
+	return publishKafkaEnvelope(ctx, topicName, bizID, envelope)
 }
 
 func handleKafkaWorkbenchEvent(ctx context.Context, envelope kafkaEnvelope, _ kafka.Message) error {
-	tenantID, err := kafkaTenantID(envelope.TenantCode)
+	tenantID, err := kafkaTenantIDFromEnvelope(envelope)
 	if err != nil {
 		return err
 	}

server/main.go

@@ -698,7 +698,7 @@ func main() {
 	startKafkaMessagePersistConsumer()
 	startKafkaFileLifecycleConsumer()
 	startKafkaWorkbenchEventConsumer()
-	startKafkaOutboxPoller()
+	startKafkaDeadLetterConsumer()
 	stopExternalBusinessEvents := startIMExternalBusinessEventPoller()
 	defer stopExternalBusinessEvents()
 

server/meetingim.conf

@@ -95,7 +95,7 @@
       "dial_timeout_ms": 2000,
       "read_timeout_ms": 2000,
       "write_timeout_ms": 2000,
-      "pending_ttl_seconds": 120,
+      "pending_ttl_seconds": 300,
       "done_ttl_seconds": 86400,
       "failed_ttl_seconds": 600,
       "hot_message_ttl_seconds": 86400,
@@ -121,7 +121,7 @@
     "sasl_mechanism": "SCRAM-SHA-512",
     "sasl_username": "kafka-external",
     "sasl_password": "gr58tNnlZOVGW3AmMbL2",
-    "max_retry_count": 5,
+    "max_retry_count": 3,
     "retry_backoff_ms": [10000, 30000, 60000, 300000, 900000],
     "dead_letter_retention_seconds": 7776000,
     "topics": {

server/store/group_message_runtime.go

@@ -51,6 +51,7 @@ type groupMessageRuntime interface {
 	MarkDone(tenantID types.TenantID, topic string, from types.Uid, clientMsgID string, seq int64, msgID string) error
 	MarkFailed(tenantID types.TenantID, topic string, from types.Uid, clientMsgID string) error
 	CacheHotMessage(tenantID types.TenantID, msg *types.Message) error
+	CachePendingHotMessage(tenantID types.TenantID, msg *types.Message) error
 	GetHotMessages(tenantID types.TenantID, topic string, opts *types.QueryOpt, maxLimit int) ([]types.Message, bool, error)
 	InvalidateHotMessages(tenantID types.TenantID, topic string, ranges []types.Range) error
 	MarkUserOnline(tenantID types.TenantID, uid types.Uid, sid, nodeID, device, userAgent string) error
@@ -187,6 +188,27 @@ func DeleteCachedObject(tenantID types.TenantID, namespace, id string) error {
 	return groupMsgRuntime.DeleteCachedObject(tenantID, namespace, id)
 }
 
+func CacheHotMessage(tenantID types.TenantID, msg *types.Message) error {
+	if groupMsgRuntime == nil || !tenantID.IsValid() || msg == nil || msg.Topic == "" || msg.SeqId <= 0 {
+		return nil
+	}
+	return groupMsgRuntime.CacheHotMessage(tenantID, msg)
+}
+
+func CachePendingHotMessage(tenantID types.TenantID, msg *types.Message) error {
+	if groupMsgRuntime == nil || !tenantID.IsValid() || msg == nil || msg.Topic == "" || msg.SeqId <= 0 {
+		return nil
+	}
+	return groupMsgRuntime.CachePendingHotMessage(tenantID, msg)
+}
+
+func GetHotMessages(tenantID types.TenantID, topic string, opts *types.QueryOpt, maxLimit int) ([]types.Message, bool, error) {
+	if groupMsgRuntime == nil || !tenantID.IsValid() || topic == "" {
+		return nil, false, nil
+	}
+	return groupMsgRuntime.GetHotMessages(tenantID, topic, opts, maxLimit)
+}
+
 func InvalidateHotMessages(tenantID types.TenantID, topic string, ranges []types.Range) error {
 	if groupMsgRuntime == nil || !tenantID.IsValid() || topic == "" {
 		return nil
@@ -239,7 +261,7 @@ func initGroupMessageRuntime(raw json.RawMessage) error {
 		cfg.KeyPrefix = "im"
 	}
 	if cfg.PendingTTL <= 0 {
-		cfg.PendingTTL = 120
+		cfg.PendingTTL = 300
 	}
 	if cfg.DoneTTL <= 0 {
 		cfg.DoneTTL = 86400
@@ -380,15 +402,25 @@ func (r *redisGroupMessageRuntime) MarkFailed(tenantID types.TenantID, topic str
 }
 
 func (r *redisGroupMessageRuntime) CacheHotMessage(tenantID types.TenantID, msg *types.Message) error {
+	return r.cacheHotMessage(tenantID, msg, time.Duration(r.cfg.HotMsgTTL)*time.Second)
+}
+
+func (r *redisGroupMessageRuntime) CachePendingHotMessage(tenantID types.TenantID, msg *types.Message) error {
+	return r.cacheHotMessage(tenantID, msg, time.Duration(r.cfg.PendingTTL)*time.Second)
+}
+
+func (r *redisGroupMessageRuntime) cacheHotMessage(tenantID types.TenantID, msg *types.Message, ttl time.Duration) error {
 	if msg == nil || msg.Topic == "" || msg.SeqId <= 0 {
 		return types.ErrMalformed
 	}
+	if ttl <= 0 {
+		return types.ErrMalformed
+	}
 	payload, err := json.Marshal(msg)
 	if err != nil {
 		return err
 	}
 	ctx := context.Background()
-	ttl := time.Duration(r.cfg.HotMsgTTL) * time.Second
 	seq := strconv.FormatInt(msg.SeqId, 10)
 	pipe := r.client.Pipeline()
 	pipe.Set(ctx, r.hotMsgKey(tenantID, msg.Topic, msg.SeqId), payload, ttl)

server/store/infrastructure.go

@@ -22,10 +22,6 @@ type HTTPInfrastructurePersistenceInterface interface {
 	GetKafkaDeadLetter(types.TenantID, string) (*types.KafkaDeadLetter, error)
 	MarkKafkaDeadLetter(types.TenantID, string, string, string, string, time.Time) error
 	MarkKafkaDeadLetterReplay(types.TenantID, string, string, string, time.Time) (*types.KafkaDeadLetter, error)
-	CreateKafkaOutbox(types.TenantID, *types.KafkaOutboxRecord) error
-	ListPendingKafkaOutbox(types.TenantID, int, time.Time) ([]types.KafkaOutboxRecord, error)
-	MarkKafkaOutboxSent(types.TenantID, string, time.Time) error
-	MarkKafkaOutboxFailed(types.TenantID, string, string, string, time.Time, time.Time) error
 }
 
 type httpInfrastructureMapper struct{}
@@ -155,35 +151,3 @@ func (httpInfrastructureMapper) MarkKafkaDeadLetterReplay(tenantID types.TenantI
 	}
 	return infrastructure.TenantKafkaDeadLetterMarkReplay(tenantID, deadID, resolvedBy, note, now)
 }
-
-func (httpInfrastructureMapper) CreateKafkaOutbox(tenantID types.TenantID, record *types.KafkaOutboxRecord) error {
-	infrastructure, err := tenantHTTPInfrastructureAdapter(tenantID)
-	if err != nil {
-		return err
-	}
-	return infrastructure.TenantKafkaOutboxCreate(tenantID, record)
-}
-
-func (httpInfrastructureMapper) ListPendingKafkaOutbox(tenantID types.TenantID, limit int, now time.Time) ([]types.KafkaOutboxRecord, error) {
-	infrastructure, err := tenantHTTPInfrastructureAdapter(tenantID)
-	if err != nil {
-		return nil, err
-	}
-	return infrastructure.TenantKafkaOutboxListPending(tenantID, limit, now)
-}
-
-func (httpInfrastructureMapper) MarkKafkaOutboxSent(tenantID types.TenantID, eventID string, now time.Time) error {
-	infrastructure, err := tenantHTTPInfrastructureAdapter(tenantID)
-	if err != nil {
-		return err
-	}
-	return infrastructure.TenantKafkaOutboxMarkSent(tenantID, eventID, now)
-}
-
-func (httpInfrastructureMapper) MarkKafkaOutboxFailed(tenantID types.TenantID, eventID, reason, message string, nextRetryAt, now time.Time) error {
-	infrastructure, err := tenantHTTPInfrastructureAdapter(tenantID)
-	if err != nil {
-		return err
-	}
-	return infrastructure.TenantKafkaOutboxMarkFailed(tenantID, eventID, reason, message, nextRetryAt, now)
-}

server/store/store.go

@@ -3443,6 +3443,10 @@ func PersistPreparedMessage(tenantID types.TenantID, msg *types.Message, attachm
 	return Messages.PersistPrepared(tenantID, msg, attachmentURLs, readBySender)
 }
 
+func NormalizeClientMsgID(head types.KVMap) (string, error) {
+	return normalizeClientMsgID(head)
+}
+
 func retryTenantMessageSaveAfterSeqConflict(tenantAdp adapter.TenantBusinessAdapter, tenantID types.TenantID, msg *types.Message) error {
 	latest, err := tenantAdp.TenantMessageGetAll(tenantID, msg.Topic, types.ZeroUid, &types.QueryOpt{Before: 1 << 62, Limit: 1})
 	if err != nil {

server/store/types/infrastructure.go

@@ -23,10 +23,6 @@ const (
 	KafkaDeadStatusReplayed      = "replayed"
 	KafkaDeadStatusIgnored       = "ignored"
 	KafkaDeadStatusResolved      = "resolved"
-
-	KafkaOutboxPending = "pending"
-	KafkaOutboxSent    = "sent"
-	KafkaOutboxFailed  = "failed"
 )
 
 // HTTPIdempotencyRecord stores the first response for one authenticated write.
@@ -115,26 +111,6 @@ type KafkaDeadLetterPage struct {
 	NextCursor string
 }
 
-// KafkaOutboxRecord stores a Kafka event waiting for reliable publication.
-type KafkaOutboxRecord struct {
-	ID          int64
-	EventID     string
-	Topic       string
-	EventType   string
-	BizType     string
-	BizID       string
-	Key         string
-	Envelope    []byte
-	Status      string
-	Attempts    int
-	NextRetryAt *time.Time
-	SentAt      *time.Time
-	LastReason  string
-	LastError   string
-	CreatedAt   time.Time
-	UpdatedAt   time.Time
-}
-
 // StableEventID derives an idempotent event identifier from domain identity.
 func StableEventID(parts ...any) string {
 	hash := sha256.New()

server/tinode.conf

@@ -95,7 +95,7 @@
       "dial_timeout_ms": 2000,
       "read_timeout_ms": 2000,
       "write_timeout_ms": 2000,
-      "pending_ttl_seconds": 120,
+      "pending_ttl_seconds": 300,
       "done_ttl_seconds": 86400,
       "failed_ttl_seconds": 600,
       "hot_message_ttl_seconds": 86400,
@@ -121,7 +121,7 @@
     "sasl_mechanism": "SCRAM-SHA-512",
     "sasl_username": "kafka-external",
     "sasl_password": "gr58tNnlZOVGW3AmMbL2",
-    "max_retry_count": 5,
+    "max_retry_count": 3,
     "retry_backoff_ms": [10000, 30000, 60000, 300000, 900000],
     "dead_letter_retention_seconds": 7776000,
     "topics": {

server/topic.go

@@ -2984,11 +2984,20 @@ func (t *Topic) replyGetData(sess *Session, asUid types.Uid, asChan bool, req *M
 	count := 0
 	if userData := t.perUser[asUid]; (userData.modeGiven & userData.modeWant).IsReader() {
 		// Read messages from DB
-		messages, err := store.Messages.GetAll(t.tenantID, t.name, asUid, msgOpts2storeOpts(req))
+		opts := msgOpts2storeOpts(req)
+		messages, err := store.Messages.GetAll(t.tenantID, t.name, asUid, opts)
 		if err != nil {
 			sess.queueOut(ErrUnknownReply(msg, now))
 			return err
 		}
+		if len(messages) == 0 && shouldTryHotMessageFallback(t.cat, t.lastID, opts) {
+			hotMessages, ok, cacheErr := store.GetHotMessages(t.tenantID, t.name, opts, hotMessageFallbackLimit(opts))
+			if cacheErr != nil {
+				logs.Warn.Printf("get.data hot message fallback failed topic=%s user=%s: %v", t.name, asUid.UserId(), cacheErr)
+			} else if ok {
+				messages = filterHotMessagesForUser(hotMessages, asUid)
+			}
+		}
 
 		// Push the list of messages to the client as {data}.
 		if messages != nil {
@@ -3032,6 +3041,56 @@ func (t *Topic) replyGetData(sess *Session, asUid types.Uid, asChan bool, req *M
 	return nil
 }
 
+func shouldTryHotMessageFallback(cat types.TopicCat, lastID int64, opts *types.QueryOpt) bool {
+	if cat != types.TopicCatGrp || opts == nil || opts.IfModifiedSince != nil ||
+		!opts.User.IsZero() || opts.Topic != "" || opts.Since <= 0 || opts.Before > 0 || len(opts.IdRanges) > 0 {
+		return false
+	}
+	limit := hotMessageFallbackLimit(opts)
+	if limit <= 0 || limit > 10 {
+		return false
+	}
+	floor := lastID - 10
+	if floor < 1 {
+		floor = 1
+	}
+	return lastID <= 0 || opts.Since >= floor
+}
+
+func hotMessageFallbackLimit(opts *types.QueryOpt) int {
+	if opts != nil && opts.Limit > 0 {
+		return opts.Limit
+	}
+	return 1
+}
+
+func filterHotMessagesForUser(messages []types.Message, asUid types.Uid) []types.Message {
+	if len(messages) == 0 {
+		return messages
+	}
+	filtered := messages[:0]
+	for i := range messages {
+		if hotMessageVisibleToUser(&messages[i], asUid) {
+			filtered = append(filtered, messages[i])
+		}
+	}
+	return filtered
+}
+
+func hotMessageVisibleToUser(msg *types.Message, asUid types.Uid) bool {
+	if msg == nil || msg.DeletedAt != nil || msg.DelId > 0 || asUid.IsZero() {
+		return msg != nil && msg.DeletedAt == nil && msg.DelId == 0
+	}
+	uid := asUid.String()
+	userID := asUid.UserId()
+	for _, deleted := range msg.DeletedFor {
+		if deleted.User == uid || deleted.User == userID {
+			return false
+		}
+	}
+	return true
+}
+
 // replyGetTags returns topics' tags - tokens used for discovery.
 func (t *Topic) replyGetTags(sess *Session, asUid types.Uid, msg *ClientComMessage) error {
 	now := types.TimeNow()

server/topic_test.go

@@ -1,6 +1,8 @@
 package main
 
 import (
+	"context"
+	"encoding/json"
 	"fmt"
 	"net/http"
 	"os"
@@ -9,6 +11,7 @@ import (
 	"time"
 
 	"github.com/golang/mock/gomock"
+	"github.com/segmentio/kafka-go"
 	"github.com/tinode/chat/server/auth"
 	"github.com/tinode/chat/server/logs"
 	"github.com/tinode/chat/server/store"
@@ -901,6 +904,196 @@ func TestHandleBroadcastDataFiltersInvalidMentions(t *testing.T) {
 	verifyResponseCodes(helper.results[0], []int{http.StatusAccepted}, t)
 }
 
+func TestSaveMessageMaybeKafkaPublishesWithoutDBPersist(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	oldKafkaBus := kafkaBus
+	oldMessages := store.Messages
+	oldPublisher := publishKafkaMessagePersist
+	defer func() {
+		kafkaBus = oldKafkaBus
+		store.Messages = oldMessages
+		publishKafkaMessagePersist = oldPublisher
+	}()
+
+	kafkaBus = &kafkaRuntime{}
+	store.Messages = mock_store.NewMockMessagesPersistenceInterface(ctrl)
+	var publishedTopic string
+	var publishedKey string
+	var publishedEnvelope kafkaEnvelope
+	publishKafkaMessagePersist = func(_ context.Context, topicName, key string, envelope kafkaEnvelope) error {
+		publishedTopic = topicName
+		publishedKey = key
+		publishedEnvelope = envelope
+		return nil
+	}
+
+	msg := &types.Message{
+		TenantID: 1,
+		Topic:    "grp-test",
+		SeqId:    1,
+		From:     types.Uid(1).String(),
+		Head:     types.KVMap{"x-im-client-mid": "client-1"},
+		Content:  "hello",
+	}
+
+	got, err := saveMessageMaybeKafka(context.Background(), 1, msg, []string{"file1"}, true)
+	if err != nil {
+		t.Fatalf("saveMessageMaybeKafka returned error: %v", err)
+	}
+	if got == nil || got.SeqId != 1 || !got.MarkedReadBySender {
+		t.Fatalf("unexpected save result: %#v", got)
+	}
+	if msg.ClientMsgID != "client-1" {
+		t.Fatalf("client message id not normalized: %q", msg.ClientMsgID)
+	}
+	if publishedTopic != "message_persist" {
+		t.Fatalf("unexpected kafka topic: %q", publishedTopic)
+	}
+	if publishedKey != "grp-test" {
+		t.Fatalf("unexpected kafka key: %q", publishedKey)
+	}
+	if publishedEnvelope.EventType != "message.persist.group" {
+		t.Fatalf("unexpected event type: %q", publishedEnvelope.EventType)
+	}
+	if publishedEnvelope.TenantID != 1 || publishedEnvelope.TenantCode != "" {
+		t.Fatalf("unexpected tenant routing fields: tenant_id=%d tenant_code=%q", publishedEnvelope.TenantID, publishedEnvelope.TenantCode)
+	}
+	if publishedEnvelope.IdempotencyKey != "client-1" {
+		t.Fatalf("unexpected idempotency key: %q", publishedEnvelope.IdempotencyKey)
+	}
+}
+
+func TestShouldTryHotMessageFallback(t *testing.T) {
+	if !shouldTryHotMessageFallback(types.TopicCatGrp, 159, &types.QueryOpt{Since: 159, Limit: 1}) {
+		t.Fatal("expected fallback for latest group seq query")
+	}
+	if shouldTryHotMessageFallback(types.TopicCatP2P, 159, &types.QueryOpt{Since: 159, Limit: 1}) {
+		t.Fatal("did not expect fallback for p2p topic")
+	}
+	if shouldTryHotMessageFallback(types.TopicCatGrp, 159, &types.QueryOpt{Since: 1, Limit: 1}) {
+		t.Fatal("did not expect fallback for old history query")
+	}
+	if shouldTryHotMessageFallback(types.TopicCatGrp, 159, &types.QueryOpt{Since: 159, Limit: 50}) {
+		t.Fatal("did not expect fallback for broad query")
+	}
+}
+
+func TestFilterHotMessagesForUserSkipsDeleted(t *testing.T) {
+	uid := types.Uid(1)
+	now := time.Now()
+	messages := []types.Message{
+		{SeqId: 1},
+		{SeqId: 2, DeletedFor: []types.SoftDelete{{User: uid.String()}}},
+		{SeqId: 3, DeletedFor: []types.SoftDelete{{User: uid.UserId()}}},
+		{SeqId: 4, DeletedAt: &now},
+		{SeqId: 5, DelId: 10},
+	}
+	filtered := filterHotMessagesForUser(messages, uid)
+	if len(filtered) != 1 || filtered[0].SeqId != 1 {
+		t.Fatalf("unexpected filtered messages: %#v", filtered)
+	}
+}
+
+func TestMarkKafkaDeadLetterPublishesBeforeDBWrite(t *testing.T) {
+	oldKafkaBus := kafkaBus
+	oldPublisher := publishKafkaDeadLetter
+	oldInfrastructure := store.HTTPInfrastructure
+	defer func() {
+		kafkaBus = oldKafkaBus
+		publishKafkaDeadLetter = oldPublisher
+		store.HTTPInfrastructure = oldInfrastructure
+	}()
+
+	kafkaBus = &kafkaRuntime{config: kafkaConfig{ConsumerGroup: "im-server"}}
+	infrastructure := &testHTTPInfrastructureStore{}
+	store.HTTPInfrastructure = infrastructure
+	var publishedEnvelope kafkaEnvelope
+	publishKafkaDeadLetter = func(_ context.Context, topicName, key string, envelope kafkaEnvelope) error {
+		if topicName != "dead_letter" || key != "event-1" {
+			t.Fatalf("unexpected dead letter publish target topic=%q key=%q", topicName, key)
+		}
+		publishedEnvelope = envelope
+		return nil
+	}
+
+	original := kafkaEnvelope{
+		EventID:   "event-1",
+		TenantID:  1,
+		EventType: "message.persist.group",
+		BizType:   "message",
+		BizID:     "grp-test:1",
+	}
+	consume := &types.KafkaConsumeRecord{
+		EventID:       "event-1",
+		ConsumerGroup: "im-server",
+		SourceTopic:   "im.message.persist",
+		RetryCount:    3,
+	}
+	if err := markKafkaDeadLetter(context.Background(), 1, original, consume, "DB_FAILED", "boom"); err != nil {
+		t.Fatalf("markKafkaDeadLetter returned error: %v", err)
+	}
+	if len(infrastructure.deadLetters) != 0 {
+		t.Fatalf("dead letter was written before dead_letter topic consumption: %#v", infrastructure.deadLetters)
+	}
+	if publishedEnvelope.EventID == "" || publishedEnvelope.TenantID != 1 {
+		t.Fatalf("dead letter event was not published correctly: %#v", publishedEnvelope)
+	}
+}
+
+func TestHandleKafkaDeadLetterWritesDB(t *testing.T) {
+	oldKafkaBus := kafkaBus
+	oldInfrastructure := store.HTTPInfrastructure
+	defer func() {
+		kafkaBus = oldKafkaBus
+		store.HTTPInfrastructure = oldInfrastructure
+	}()
+
+	kafkaBus = nil
+	infrastructure := &testHTTPInfrastructureStore{}
+	store.HTTPInfrastructure = infrastructure
+	original := kafkaEnvelope{
+		EventID:   "event-1",
+		TenantID:  1,
+		EventType: "message.persist.group",
+		BizType:   "message",
+		BizID:     "grp-test:1",
+	}
+	encodedOriginal, _ := json.Marshal(original)
+	dead := types.KafkaDeadLetter{
+		DeadID:        "dead-1",
+		EventID:       "event-1",
+		SourceTopic:   "im.message.persist",
+		ConsumerGroup: "im-server",
+		EventType:     "message.persist.group",
+		BizType:       "message",
+		BizID:         "grp-test:1",
+		Envelope:      encodedOriginal,
+		RetryCount:    3,
+		LastReason:    "DB_FAILED",
+		LastError:     "boom",
+		Status:        types.KafkaDeadStatusDead,
+	}
+	payload, _ := json.Marshal(dead)
+	envelope := kafkaEnvelope{
+		EventID:       "dead-1",
+		TenantID:      1,
+		EventType:     "kafka.dead_letter.created",
+		ParentEventID: "event-1",
+		Payload:       payload,
+	}
+	if err := handleKafkaDeadLetter(context.Background(), envelope, kafka.Message{}); err != nil {
+		t.Fatalf("handleKafkaDeadLetter returned error: %v", err)
+	}
+	if len(infrastructure.deadLetters) != 1 {
+		t.Fatalf("expected one dead letter write, got %d", len(infrastructure.deadLetters))
+	}
+	if infrastructure.deadLetters[0].DeadID != "dead-1" || infrastructure.deadLetters[0].EventID != "event-1" {
+		t.Fatalf("unexpected dead letter write: %#v", infrastructure.deadLetters[0])
+	}
+}
+
 func TestHandleBroadcastDataInactiveTopic(t *testing.T) {
 	numUsers := 2
 	helper := TopicTestHelper{}