im_server/server/internal_api.go

package main

import (
	"bytes"
	"crypto/hmac"
	"crypto/sha256"
	"encoding/base64"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"runtime/debug"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/tinode/chat/server/auth"
	"github.com/tinode/chat/server/logs"
	"github.com/tinode/chat/server/store"
	"github.com/tinode/chat/server/store/types"
)

const (
	internalCapabilityAdminQuery = "admin_query"
	internalCapabilityCultivate  = "cultivate"

	internalHeaderService   = "X-Service-Name"
	internalHeaderTimestamp = "X-Timestamp"
	internalHeaderNonce     = "X-Nonce"
	internalHeaderSignature = "X-Signature"
)

type internalAPIConfig struct {
	Enabled           bool                                `json:"enabled"`
	TimeWindowSeconds int                                 `json:"time_window_seconds"`
	Services          map[string]internalAPIServiceConfig `json:"services"`
}

type internalAPIServiceConfig struct {
	Secret string   `json:"secret"`
	Allow  []string `json:"allow"`
}

type internalAPIRuntime struct {
	config internalAPIConfig
	nonces sync.Map
}

var internalAPI internalAPIRuntime

func initInternalAPI(raw json.RawMessage) error {
	internalAPI = internalAPIRuntime{}
	if len(raw) == 0 || string(raw) == "null" {
		return nil
	}
	var cfg internalAPIConfig
	if err := json.Unmarshal(raw, &cfg); err != nil {
		return err
	}
	if cfg.TimeWindowSeconds <= 0 {
		cfg.TimeWindowSeconds = 300
	}
	if cfg.Services == nil {
		cfg.Services = map[string]internalAPIServiceConfig{}
	}
	internalAPI.config = cfg
	if cfg.Enabled {
		logs.Info.Printf("Internal API enabled for %d service(s)", len(cfg.Services))
	}
	return nil
}

func serveInternalAdminQuery(wrt http.ResponseWriter, req *http.Request, prefix string) {
	defer func() {
		if recovered := recover(); recovered != nil {
			logs.Err.Printf("internal admin query panic path:%s err:%v stack:%s", req.URL.RequestURI(), recovered, debug.Stack())
			writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil)
		}
	}()
	if req.Method != http.MethodGet && req.Method != http.MethodPost {
		writeIMError(wrt, req, http.StatusMethodNotAllowed, 30002, imReasonMethodNotAllowed, "method not allowed", nil)
		return
	}
	if !authorizeInternalAPI(wrt, req, internalCapabilityAdminQuery) {
		return
	}

	tail := strings.Trim(strings.TrimPrefix(req.URL.Path, prefix), "/")
	parts := splitInternalPath(tail)
	if len(parts) < 3 || parts[0] != "tenants" {
		writeIMError(wrt, req, http.StatusNotFound, 30004, imReasonMalformedRequest, "resource not found", nil)
		return
	}
	tenant, ok := resolveInternalTenant(wrt, req, parts[1])
	if !ok {
		return
	}
	if req.Method == http.MethodPost && !isInternalAdminPostRoute(parts) {
		writeIMError(wrt, req, http.StatusMethodNotAllowed, 30002, imReasonMethodNotAllowed, "method not allowed", nil)
		return
	}

	switch {
	case len(parts) == 4 && parts[2] == "dashboard" && parts[3] == "overview":
		serveInternalDashboardOverview(wrt, req, tenant.ID)
	case len(parts) == 4 && parts[2] == "dashboard" && parts[3] == "trends":
		serveInternalDashboardTrends(wrt, req, tenant.ID)
	case len(parts) == 3 && parts[2] == "private-messages":
		serveInternalAdminResourceList(wrt, req, tenant.ID, "private-messages", "")
	case len(parts) == 3 && parts[2] == "group-messages":
		serveInternalAdminResourceList(wrt, req, tenant.ID, "group-messages", "")
	case req.Method == http.MethodPost && len(parts) == 3 && parts[2] == "users":
		serveInternalUserCreate(wrt, req, tenant.ID)
	case req.Method == http.MethodPost && len(parts) == 5 && parts[2] == "users" && parts[4] == "action":
		serveInternalUserAction(wrt, req, tenant.ID, parts[3])
	case req.Method == http.MethodPost && len(parts) == 4 && parts[2] == "login-lockouts" && parts[3] == "release":
		serveInternalLoginLockoutRelease(wrt, req, tenant.ID)
	case req.Method == http.MethodGet && len(parts) == 3 && parts[2] == "users":
		serveInternalAdminResourceList(wrt, req, tenant.ID, "users", "")
	case len(parts) == 4 && parts[2] == "users":
		serveInternalUserQuery(wrt, req, tenant.ID, parts[3])
	case len(parts) == 5 && parts[2] == "users" && parts[4] == "devices":
		serveInternalAdminResourceList(wrt, req, tenant.ID, "user_devices", parts[3])
	case len(parts) == 3 && parts[2] == "groups":
		serveInternalAdminResourceList(wrt, req, tenant.ID, "groups", "")
	case len(parts) == 4 && parts[2] == "groups":
		serveInternalGroupQuery(wrt, req, tenant.ID, parts[3])
	case len(parts) == 5 && parts[2] == "groups" && parts[4] == "members":
		serveInternalGroupMembersQuery(wrt, req, tenant.ID, parts[3])
	case len(parts) == 5 && parts[2] == "groups" && internalGroupResourceSpecs[parts[4]] != nil:
		serveInternalAdminResourceList(wrt, req, tenant.ID, "group_"+parts[4], parts[3])
	case len(parts) == 5 && parts[2] == "messages":
		serveInternalMessageQuery(wrt, req, tenant.ID, parts[3], parts[4])
	case len(parts) == 3 && parts[2] == "files":
		serveInternalAdminResourceList(wrt, req, tenant.ID, "files", "")
	case len(parts) == 4 && parts[2] == "files":
		serveInternalFileQuery(wrt, req, tenant.ID, parts[3])
	case len(parts) == 5 && parts[2] == "files" && parts[4] == "links":
		serveInternalAdminResourceList(wrt, req, tenant.ID, "file_links", parts[3])
	case len(parts) == 3 && internalTenantResourceSpecs[parts[2]] != nil:
		serveInternalAdminResourceList(wrt, req, tenant.ID, parts[2], "")
	case len(parts) == 4 && internalTenantResourceSpecs[parts[2]] != nil:
		serveInternalAdminResourceDetail(wrt, req, tenant.ID, parts[2], parts[3])
	case len(parts) == 4 && parts[2] == "reports" && parts[3] == "abuse":
		writeIMError(wrt, req, http.StatusNotFound, 30004, imReasonMalformedRequest, "resource not found", nil)
	case len(parts) == 6 && parts[2] == "reports" && parts[3] == "abuse" && parts[5] == "target":
		serveInternalAbuseTargetQuery(wrt, req, tenant.ID, parts[4])
	case len(parts) == 4 && parts[2] == "outbox" && internalOutboxResourceSpecs[parts[3]] != nil:
		serveInternalAdminResourceList(wrt, req, tenant.ID, "outbox_"+parts[3], "")
	case len(parts) == 5 && parts[2] == "outbox" && internalOutboxResourceSpecs[parts[3]] != nil:
		serveInternalAdminResourceDetail(wrt, req, tenant.ID, "outbox_"+parts[3], parts[4])
	case len(parts) == 4 && parts[2] == "dashboard" && parts[3] == "summary":
		serveInternalDashboardSummary(wrt, req, tenant.ID)
	case len(parts) == 4 && parts[2] == "kafka" && parts[3] == "dead-letters":
		serveInternalKafkaDeadLettersQuery(wrt, req, tenant.ID)
	case len(parts) == 5 && parts[2] == "kafka" && parts[3] == "dead-letters":
		serveInternalKafkaDeadLetterQuery(wrt, req, tenant.ID, parts[4])
	default:
		writeIMError(wrt, req, http.StatusNotFound, 30004, imReasonMalformedRequest, "resource not found", nil)
	}
}

func isInternalAdminPostRoute(parts []string) bool {
	return (len(parts) == 3 && parts[2] == "users") ||
		(len(parts) == 5 && parts[2] == "users" && parts[4] == "action") ||
		(len(parts) == 4 && parts[2] == "login-lockouts" && parts[3] == "release")
}

type internalUserCreateRequest struct {
	Username string `json:"username"`
	Nickname string `json:"nickname"`
	Phone    string `json:"phone"`
	Password string `json:"password"`
}

type internalUserActionRequest struct {
	EventType    string `json:"event_type"`
	ActorUserID  string `json:"actor_user_id"`
	DeviceID     string `json:"device_id"`
	PasswordHash string `json:"password_hash"`
	Reason       string `json:"reason"`
	Hard         bool   `json:"hard"`
}

func serveInternalUserCreate(wrt http.ResponseWriter, req *http.Request, tenantID types.TenantID) {
	var body internalUserCreateRequest
	if err := json.NewDecoder(io.LimitReader(req.Body, 1<<20)).Decode(&body); err != nil {
		writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", nil)
		return
	}
	body.Username = strings.ToLower(strings.TrimSpace(body.Username))
	body.Nickname = strings.TrimSpace(body.Nickname)
	body.Phone = strings.TrimSpace(body.Phone)
	if body.Username == "" {
		var err error
		body.Username, err = randomRegistrationUsername()
		if err != nil {
			writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil)
			return
		}
	}
	if err := validateRegistrationUsername(body.Username); err != nil || len([]rune(body.Password)) < 3 || len([]byte(body.Password)) > 72 {
		writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "用户名或密码格式不正确", nil)
		return
	}
	if body.Phone != "" {
		phone, err := normalizeIMPhone(body.Phone, "CN")
		if err != nil {
			writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "手机号格式不正确", nil)
			return
		}
		body.Phone = phone
	}
	authHandler := store.Store.GetLogicalAuthHandler("basic")
	if authHandler == nil {
		writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "basic authentication unavailable", nil)
		return
	}
	authCtx := auth.AuthContext{TenantID: tenantID, RemoteAddr: req.RemoteAddr}
	secret := []byte(body.Username + ":" + body.Password)
	if unique, err := authHandler.IsUnique(authCtx, secret); err != nil || !unique {
		writeIMError(wrt, req, http.StatusConflict, 30001, "DUPLICATE_USER", "用户名已存在", nil)
		return
	}
	user := &types.User{
		TenantID: tenantID,
		State:    types.StateOK,
		Username: body.Username,
		Phone:    body.Phone,
		Public:   map[string]any{"fn": firstNonEmpty(body.Nickname, body.Username)},
	}
	user.Access.Auth = getDefaultAccess(types.TopicCatP2P, true, false) | getDefaultAccess(types.TopicCatGrp, true, false)
	user.Access.Anon = getDefaultAccess(types.TopicCatP2P, false, false) | getDefaultAccess(types.TopicCatGrp, false, false)
	created, err := store.Users.Create(tenantID, user, nil)
	if err != nil {
		writeInternalQueryError(wrt, req, err)
		return
	}
	if _, err = authHandler.AddRecord(authCtx, &auth.Rec{TenantID: tenantID, Uid: created.Uid()}, secret); err != nil {
		_ = store.Users.Delete(tenantID, created.Uid(), true)
		writeInternalQueryError(wrt, req, err)
		return
	}
	displayNo := int64(0)
	if row, err := store.InternalAdminQueries.QueryOne(tenantID, "SELECT account_no FROM im_users WHERE id=?", []any{store.DecodeUid(created.Uid())}); err == nil {
		displayNo, _ = numericID(row["account_no"])
	} else {
		logs.Warn.Printf("internal user create account number query failed tenant:%d user:%d err:%v", tenantID, store.DecodeUid(created.Uid()), err)
	}
	writeIMSuccess(wrt, req, map[string]any{
		"user_id":    store.DecodeUid(created.Uid()),
		"display_no": displayNo,
		"username":   created.Username,
		"nickname":   body.Nickname,
		"phone":      created.Phone,
	})
}

func serveInternalUserAction(wrt http.ResponseWriter, req *http.Request, tenantID types.TenantID, rawUserID string) {
	var body internalUserActionRequest
	if err := json.NewDecoder(io.LimitReader(req.Body, 1<<20)).Decode(&body); err != nil {
		writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", nil)
		return
	}
	eventType := strings.TrimSpace(body.EventType)
	if !isInternalUserActionSupported(eventType) {
		writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "unsupported user action", map[string]any{"event_type": eventType})
		return
	}
	target := parseKafkaCommandUser(rawUserID)
	if target.IsZero() {
		writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "用户ID不能为空", nil)
		return
	}
	payload := adminCommandPayload{
		CommandID:    firstNonEmpty(strings.TrimSpace(req.Header.Get("X-Request-ID")), "internal-"+strconv.FormatInt(time.Now().UTC().UnixNano(), 10)),
		ActorUserID:  strings.TrimSpace(body.ActorUserID),
		UserID:       strconv.FormatInt(internalAdminUIDRaw(target), 10),
		TargetID:     strconv.FormatInt(internalAdminUIDRaw(target), 10),
		DeviceID:     strings.TrimSpace(body.DeviceID),
		PasswordHash: strings.TrimSpace(body.PasswordHash),
		Reason:       strings.TrimSpace(body.Reason),
		Hard:         body.Hard,
	}
	envelope := kafkaEnvelope{
		EventID:    payload.CommandID,
		TraceID:    strings.TrimSpace(req.Header.Get("X-Request-ID")),
		EventType:  eventType,
		BizType:    "user",
		BizID:      strconv.FormatInt(internalAdminUIDRaw(target), 10),
		OccurredAt: time.Now().UTC(),
	}
	var (
		result map[string]any
		err    error
	)
	if eventType == adminCommandUserPasswordReset {
		result, err = executeKafkaUserPasswordResetCommand(tenantID, payload)
	} else {
		result, err = executeKafkaUserGovernanceCommand(tenantID, envelope, payload, parseKafkaCommandUser(payload.ActorUserID))
	}
	if err != nil {
		writeInternalQueryError(wrt, req, err)
		return
	}
	if err := enqueueKafkaAuditEvent(req.Context(), tenantID, "admin", parseKafkaCommandUser(payload.ActorUserID),
		eventType, "internal_user_action", payload.CommandID, "succeeded", "", result); err != nil {
		logs.Warn.Printf("internal user action audit enqueue failed tenant:%d user:%d event:%s err:%v", tenantID, internalAdminUIDRaw(target), eventType, err)
	}
	writeIMSuccess(wrt, req, map[string]any{
		"user_id":    internalAdminUIDRaw(target),
		"event_type": eventType,
		"status":     "succeeded",
		"result":     result,
	})
}

func isInternalUserActionSupported(eventType string) bool {
	switch strings.TrimSpace(eventType) {
	case adminCommandUserBan, adminCommandUserUnban,
		adminCommandUserFreeze, adminCommandUserUnfreeze,
		adminCommandUserDeviceKick, adminCommandUserLogoutAll,
		adminCommandUserCancel, adminCommandUserDelete,
		adminCommandUserPasswordReset:
		return true
	default:
		return false
	}
}

type internalLoginLockoutReleaseRequest struct {
	AccountHash string `json:"account_hash"`
}

// serveInternalLoginLockoutRelease clears the login failure lockout for one
// account so the user can sign in again immediately.
func serveInternalLoginLockoutRelease(wrt http.ResponseWriter, req *http.Request, tenantID types.TenantID) {
	var body internalLoginLockoutReleaseRequest
	if err := json.NewDecoder(io.LimitReader(req.Body, 1<<20)).Decode(&body); err != nil {
		writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", nil)
		return
	}
	accountHash := strings.ToLower(strings.TrimSpace(body.AccountHash))
	if len(accountHash) != 64 {
		writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "account_hash 格式不正确", nil)
		return
	}
	rec, err := store.LoginLockouts.Get(tenantID, accountHash)
	if err != nil && !errors.Is(err, types.ErrNotFound) && !errors.Is(err, types.ErrUnsupported) {
		writeInternalQueryError(wrt, req, err)
		return
	}
	if err = store.LoginLockouts.Release(tenantID, accountHash, time.Now().UTC().Round(time.Millisecond)); err != nil {
		writeInternalQueryError(wrt, req, err)
		return
	}
	logs.Info.Printf("login lockout released tenant:%d account_hash:%s had_lock:%v", tenantID, accountHash,
		rec != nil && rec.LockedUntil != nil)
	writeIMSuccess(wrt, req, map[string]any{"account_hash": accountHash, "status": "released"})
}

func authorizeInternalAPI(wrt http.ResponseWriter, req *http.Request, capability string) bool {
	if !internalAPI.config.Enabled {
		writeIMError(wrt, req, http.StatusNotFound, 30004, imReasonMalformedRequest, "resource not found", nil)
		return false
	}
	serviceName := strings.TrimSpace(req.Header.Get(internalHeaderService))
	timestampRaw := strings.TrimSpace(req.Header.Get(internalHeaderTimestamp))
	nonce := strings.TrimSpace(req.Header.Get(internalHeaderNonce))
	signature := strings.TrimSpace(req.Header.Get(internalHeaderSignature))
	service, ok := internalAPI.config.Services[serviceName]
	if !ok || service.Secret == "" || timestampRaw == "" || nonce == "" || signature == "" {
		writeIMError(wrt, req, http.StatusUnauthorized, 30043, imReasonAuthRequired, "internal signature required", nil)
		return false
	}
	if !internalServiceAllows(service, capability) {
		writeIMError(wrt, req, http.StatusForbidden, 30005, imReasonPermissionDenied, "permission denied", nil)
		return false
	}
	timestamp, err := strconv.ParseInt(timestampRaw, 10, 64)
	if err != nil {
		writeIMError(wrt, req, http.StatusUnauthorized, 30043, imReasonAuthRequired, "invalid internal timestamp", nil)
		return false
	}
	window := time.Duration(internalAPI.config.TimeWindowSeconds) * time.Second
	if delta := time.Since(time.Unix(timestamp, 0)); delta > window || delta < -window {
		writeIMError(wrt, req, http.StatusUnauthorized, 30043, imReasonAuthRequired, "internal signature expired", nil)
		return false
	}
	bodyHash, err := hashInternalRequestBody(req)
	if err != nil {
		writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil)
		return false
	}
	signingText := strings.Join([]string{
		req.Method,
		req.URL.RequestURI(),
		bodyHash,
		timestampRaw,
		nonce,
	}, "\n")
	if !verifyInternalSignature(service.Secret, signingText, signature) {
		writeIMError(wrt, req, http.StatusUnauthorized, 30043, imReasonAuthRequired, "invalid internal signature", nil)
		return false
	}
	requestClaim := internalRequestClaimValue(signingText)
	if !claimInternalNonce(serviceName, nonce, requestClaim, time.Now().Add(window)) {
		writeIMError(wrt, req, http.StatusUnauthorized, 30043, imReasonAuthRequired, "internal nonce replayed", nil)
		return false
	}
	return true
}

func splitInternalPath(tail string) []string {
	if tail == "" {
		return nil
	}
	rawParts := strings.Split(tail, "/")
	parts := make([]string, 0, len(rawParts))
	for _, part := range rawParts {
		if part == "" {
			continue
		}
		decoded, err := url.PathUnescape(part)
		if err != nil {
			decoded = part
		}
		parts = append(parts, decoded)
	}
	return parts
}

func resolveInternalTenant(wrt http.ResponseWriter, req *http.Request, tenantCode string) (*types.Tenant, bool) {
	if tenantCode == "" || store.Tenants == nil {
		writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "tenant_code"})
		return nil, false
	}
	tenant, err := store.Tenants.GetByCode(tenantCode)
	if err != nil {
		if errors.Is(err, types.ErrNotFound) {
			writeIMError(wrt, req, http.StatusNotFound, 30010, imReasonTenantUnavailable, "tenant unavailable", nil)
			return nil, false
		}
		writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil)
		return nil, false
	}
	if !tenant.IsActive() {
		writeIMError(wrt, req, http.StatusForbidden, 30010, imReasonTenantUnavailable, "tenant unavailable", nil)
		return nil, false
	}
	return tenant, true
}

func serveInternalUserQuery(wrt http.ResponseWriter, req *http.Request, tenantID types.TenantID, userID string) {
	userDBID, ok := parseInternalAdminDBID(userID)
	if !ok {
		writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "user_id"})
		return
	}
	if spec, ok := internalAdminQuerySpec("users"); ok {
		query := spec.Select + " " + spec.From + " WHERE u.id=?"
		if row, err := store.InternalAdminQueries.QueryOne(tenantID, query, []any{userDBID}); err == nil {
			writeIMSuccess(wrt, req, normalizeInternalAdminRowForAPI(row))
			return
		} else {
			logs.Warn.Printf("internal user query sql failed tenant:%d user:%d err:%v", tenantID, userDBID, err)
		}
	}
	uid := store.EncodeUid(userDBID)
	user, err := store.Users.Get(tenantID, uid)
	if err != nil {
		logs.Warn.Printf("internal user query fallback failed tenant:%d user:%d err:%v", tenantID, userDBID, err)
		writeInternalQueryError(wrt, req, err)
		return
	}
	if user == nil {
		writeInternalQueryError(wrt, req, types.ErrUserNotFound)
		return
	}
	writeIMSuccess(wrt, req, map[string]any{
		"user_id":    userDBID,
		"state":      user.State,
		"state_at":   user.StateAt,
		"username":   user.Username,
		"phone":      user.Phone,
		"public":     user.Public,
		"trusted":    user.Trusted,
		"last_seen":  user.LastSeen,
		"user_agent": user.UserAgent,
		"tags":       user.Tags,
		"created_at": user.CreatedAt,
		"updated_at": user.UpdatedAt,
	})
}

func serveInternalGroupQuery(wrt http.ResponseWriter, req *http.Request, tenantID types.TenantID, topic string) {
	if topic == "" || store.GroupBusiness == nil {
		writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "topic"})
		return
	}
	if spec, ok := internalAdminQuerySpec("groups"); ok {
		query := spec.Select + " " + spec.From + " WHERE t.name=?"
		if row, err := store.InternalAdminQueries.QueryOne(tenantID, query, []any{topic}); err == nil {
			writeIMSuccess(wrt, req, normalizeInternalAdminRowForAPI(row))
			return
		}
	}
	settings, err := store.GroupBusiness.GetSettings(tenantID, topic)
	if err != nil {
		writeInternalQueryError(wrt, req, err)
		return
	}
	writeIMSuccess(wrt, req, internalGroupSettingsPayload(settings))
}

func serveInternalGroupMembersQuery(wrt http.ResponseWriter, req *http.Request, tenantID types.TenantID, topic string) {
	if topic == "" || store.GroupBusiness == nil {
		writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "topic"})
		return
	}
	spec := &internalQuerySpec{
		Select: `SELECT gm.id,gm.topic,gm.user_id,u.state,u.public,p.nickname,p.public_account,s.modewant AS mode_want,s.modegiven AS mode_given,s.recvseqid AS recv_seq_id,s.readseqid AS read_seq_id,s.delid AS del_id,
			gm.role,gm.member_status,gm.nickname AS group_nickname,gm.pinned,gm.mute_notifications,gm.join_source,gm.joined_by,gm.joined_at,gm.left_at,gm.removed_by,gm.version,gm.created_at,gm.updated_at`,
		From:        "FROM im_group_members AS gm LEFT JOIN im_users AS u ON u.id=gm.user_id LEFT JOIN im_user_profiles AS p ON p.user_id=gm.user_id LEFT JOIN im_subscriptions AS s ON s.topic=gm.topic AND s.userid=gm.user_id",
		DefaultSort: "ORDER BY gm.id DESC",
		ScopeColumn: "gm.topic",
		Filters: []internalQueryFilter{
			{"user_id", "gm.user_id", "=", "uid"},
			{"role", "gm.role", "=", "string"},
			{"member_status", "gm.member_status", "=", "string"},
			{"q", "gm.nickname", "LIKE", "string"},
			{"eligible_role", "gm.role", "=", "string"},
		},
	}
	query, countQuery, args, page, perPage, err := buildInternalAdminListQuery(req, spec, topic)
	if err == nil {
		if result, queryErr := store.InternalAdminQueries.Query(tenantID, query, countQuery, args, page, perPage); queryErr == nil {
			writeIMSuccess(wrt, req, map[string]any{
				"items":       normalizeInternalAdminRowsForAPI(result.Items),
				"total":       result.Total,
				"page":        result.Page,
				"per_page":    result.PerPage,
				"next_cursor": "",
				"has_more":    false,
			})
			return
		}
	}
	limit := parseIMLimit(req.URL.Query().Get("limit"), 50, 200)
	members, nextCursor, err := store.GroupBusiness.ListMembers(tenantID, topic, types.GroupMemberListOptions{
		Role:         strings.TrimSpace(req.URL.Query().Get("role")),
		EligibleRole: strings.TrimSpace(req.URL.Query().Get("eligible_role")),
		Query:        strings.TrimSpace(req.URL.Query().Get("q")),
		Cursor:       strings.TrimSpace(req.URL.Query().Get("cursor")),
		Limit:        limit,
	})
	if err != nil {
		writeInternalQueryError(wrt, req, err)
		return
	}
	items := make([]any, 0, len(members))
	for _, member := range members {
		items = append(items, map[string]any{
			"user_id":            internalAdminUIDRaw(member.UserID),
			"role":               member.Role,
			"member_status":      member.MemberStatus,
			"nickname":           member.Nickname,
			"pinned":             member.Pinned,
			"mute_notifications": member.MuteNotifications,
			"join_source":        member.JoinSource,
			"joined_by":          internalAdminUIDRaw(member.JoinedBy),
			"joined_at":          member.JoinedAt,
			"left_at":            member.LeftAt,
			"removed_by":         internalAdminUIDRaw(member.RemovedBy),
			"version":            member.Version,
			"created_at":         member.CreatedAt,
			"updated_at":         member.UpdatedAt,
		})
	}
	writeIMSuccess(wrt, req, map[string]any{"items": items, "next_cursor": nextCursor})
}

func serveInternalMessageQuery(wrt http.ResponseWriter, req *http.Request, tenantID types.TenantID, topic, seqRaw string) {
	seq, err := strconv.ParseInt(seqRaw, 10, 64)
	if topic == "" || err != nil || seq <= 0 {
		writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", nil)
		return
	}
	resource := "private-messages"
	spec := privateMessagesInternalQuerySpec()
	where := "m.topic=? AND m.seqid=?"
	if strings.HasPrefix(topic, "grp") {
		resource = "group-messages"
		spec, err = groupMessagesInternalQuerySpecForTenant(tenantID)
		if err != nil {
			writeInternalQueryError(wrt, req, err)
			return
		}
	}
	spec.Select += ",m.head,JSON_UNQUOTE(m.content) AS content"
	row, err := store.InternalAdminQueries.QueryOne(tenantID, spec.Select+" "+spec.From+" WHERE "+where, []any{topic, seq})
	if err != nil {
		writeInternalQueryError(wrt, req, err)
		return
	}
	row["conversation_type"] = strings.TrimSuffix(resource, "-messages")
	writeIMSuccess(wrt, req, normalizeInternalAdminRowForAPI(row))
}

func serveInternalFileQuery(wrt http.ResponseWriter, req *http.Request, tenantID types.TenantID, fileID string) {
	if fileID == "" {
		writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "file_id"})
		return
	}
	fileDBID, ok := parseInternalAdminDBID(fileID)
	if !ok {
		writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "file_id"})
		return
	}
	spec, _ := internalAdminQuerySpec("files")
	row, err := store.InternalAdminQueries.QueryOne(tenantID, spec.Select+" "+spec.From+" WHERE id=?", []any{fileDBID})
	if err != nil {
		writeInternalQueryError(wrt, req, err)
		return
	}
	writeIMSuccess(wrt, req, normalizeInternalAdminRowForAPI(row))
}

func serveInternalKafkaDeadLettersQuery(wrt http.ResponseWriter, req *http.Request, tenantID types.TenantID) {
	if store.HTTPInfrastructure == nil {
		writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil)
		return
	}
	limit := parseIMLimit(req.URL.Query().Get("limit"), 50, 200)
	page, err := store.HTTPInfrastructure.ListKafkaDeadLetters(tenantID, types.KafkaDeadLetterListOptions{
		Status:    strings.TrimSpace(req.URL.Query().Get("status")),
		BizType:   strings.TrimSpace(req.URL.Query().Get("biz_type")),
		BizID:     strings.TrimSpace(req.URL.Query().Get("biz_id")),
		EventType: strings.TrimSpace(req.URL.Query().Get("event_type")),
		Cursor:    strings.TrimSpace(req.URL.Query().Get("cursor")),
		Limit:     limit,
	})
	if err != nil {
		writeInternalQueryError(wrt, req, err)
		return
	}
	items := make([]any, 0, len(page.Items))
	for _, dead := range page.Items {
		items = append(items, internalKafkaDeadLetterPayload(&dead, false))
	}
	writeIMSuccess(wrt, req, map[string]any{
		"items":       items,
		"next_cursor": page.NextCursor,
		"has_more":    page.NextCursor != "",
	})
}

func serveInternalKafkaDeadLetterQuery(wrt http.ResponseWriter, req *http.Request, tenantID types.TenantID, deadID string) {
	if store.HTTPInfrastructure == nil || strings.TrimSpace(deadID) == "" {
		writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "dead_id"})
		return
	}
	dead, err := store.HTTPInfrastructure.GetKafkaDeadLetter(tenantID, deadID)
	if err != nil {
		writeInternalQueryError(wrt, req, err)
		return
	}
	writeIMSuccess(wrt, req, internalKafkaDeadLetterPayload(dead, true))
}

func internalGroupSettingsPayload(settings *types.GroupSettings) map[string]any {
	if settings == nil {
		return nil
	}
	return map[string]any{
		"topic":               settings.Topic,
		"owner_user_id":       internalAdminUIDRaw(settings.OwnerUserID),
		"name":                settings.Name,
		"avatar_file_id":      settings.AvatarFileID,
		"description":         settings.Description,
		"group_status":        settings.GroupStatus,
		"join_approval":       settings.JoinApproval,
		"mute_all":            settings.MuteAll,
		"member_upload_files": settings.MemberUploadFiles,
		"member_private_chat": settings.MemberPrivateChat,
		"member_invite":       settings.MemberInvite,
		"max_member_count":    settings.MaxMemberCount,
		"max_admin_count":     settings.MaxAdminCount,
		"member_count":        settings.MemberCount,
		"version":             settings.Version,
		"created_by":          internalAdminUIDRaw(settings.CreatedBy),
		"dissolved_by":        internalAdminUIDRaw(settings.DissolvedBy),
		"dissolved_at":        settings.DissolvedAt,
		"created_at":          settings.CreatedAt,
		"updated_at":          settings.UpdatedAt,
	}
}

func internalKafkaDeadLetterPayload(dead *types.KafkaDeadLetter, includeEnvelope bool) map[string]any {
	if dead == nil {
		return nil
	}
	payload := map[string]any{
		"id":              dead.ID,
		"dead_id":         dead.DeadID,
		"event_id":        dead.EventID,
		"parent_event_id": dead.ParentEventID,
		"source_topic":    dead.SourceTopic,
		"consumer_group":  dead.ConsumerGroup,
		"event_type":      dead.EventType,
		"biz_type":        dead.BizType,
		"biz_id":          dead.BizID,
		"retry_count":     dead.RetryCount,
		"last_reason":     dead.LastReason,
		"last_error":      dead.LastError,
		"status":          dead.Status,
		"replay_count":    dead.ReplayCount,
		"created_at":      dead.CreatedAt,
		"updated_at":      dead.UpdatedAt,
	}
	if includeEnvelope {
		var envelope any
		if len(dead.Envelope) > 0 && json.Unmarshal(dead.Envelope, &envelope) == nil {
			payload["envelope"] = envelope
		} else {
			payload["envelope_raw"] = string(dead.Envelope)
		}
	}
	return payload
}

func writeInternalQueryError(wrt http.ResponseWriter, req *http.Request, err error) {
	logs.Warn.Printf("internal admin query failed: method=%s path=%s err=%v", req.Method, req.URL.RequestURI(), err)
	switch {
	case errors.Is(err, types.ErrUserNotFound), errors.Is(err, types.ErrTopicNotFound), errors.Is(err, types.ErrNotFound):
		writeIMError(wrt, req, http.StatusNotFound, 30004, imReasonMalformedRequest, "resource not found", nil)
	case errors.Is(err, types.ErrMalformed):
		writeIMError(wrt, req, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", nil)
	case errors.Is(err, types.ErrPermissionDenied):
		writeIMError(wrt, req, http.StatusForbidden, 30005, imReasonPermissionDenied, "permission denied", nil)
	default:
		writeIMError(wrt, req, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil)
	}
}

func parseInternalUid(raw string) types.Uid {
	if uid := types.ParseUserId(raw); !uid.IsZero() {
		return uid
	}
	id, ok := parseInternalAdminDBID(raw)
	if !ok {
		return types.ZeroUid
	}
	return store.EncodeUid(id)
}

func internalServiceAllows(service internalAPIServiceConfig, capability string) bool {
	for _, allowed := range service.Allow {
		allowed = strings.TrimSpace(allowed)
		if allowed == "*" || allowed == capability {
			return true
		}
	}
	return false
}

func hashInternalRequestBody(req *http.Request) (string, error) {
	if req.Body == nil {
		sum := sha256.Sum256(nil)
		return hex.EncodeToString(sum[:]), nil
	}
	body, err := io.ReadAll(io.LimitReader(req.Body, 1<<20))
	if err != nil {
		return "", err
	}
	if err := req.Body.Close(); err != nil {
		return "", err
	}
	req.Body = io.NopCloser(bytes.NewReader(body))
	sum := sha256.Sum256(body)
	return hex.EncodeToString(sum[:]), nil
}

func verifyInternalSignature(secret, signingText, signature string) bool {
	signature = strings.TrimPrefix(signature, "sha256=")
	mac := hmac.New(sha256.New, []byte(secret))
	_, _ = mac.Write([]byte(signingText))
	expected := mac.Sum(nil)
	if decoded, err := hex.DecodeString(signature); err == nil {
		return hmac.Equal(decoded, expected)
	}
	if decoded, err := base64.StdEncoding.DecodeString(signature); err == nil {
		return hmac.Equal(decoded, expected)
	}
	return false
}

func claimInternalNonce(serviceName, nonce, requestClaim string, expiresAt time.Time) bool {
	ttl := time.Until(expiresAt)
	nonceKey := fmt.Sprintf("%s:%s", serviceName, nonce)
	if claimed, _, enabled, err := store.ClaimShortStateValue("internal_api_nonce", nonceKey, requestClaim, ttl); enabled {
		if err != nil {
			logs.Warn.Println("internal api: failed to claim nonce in redis", err)
			return false
		}
		return claimed
	}
	now := time.Now()
	internalAPI.nonces.Range(func(key, value any) bool {
		if claim, ok := value.(internalNonceClaim); ok && claim.expiresAt.Before(now) {
			internalAPI.nonces.Delete(key)
		}
		return true
	})
	if _, loaded := internalAPI.nonces.LoadOrStore(nonceKey, internalNonceClaim{value: requestClaim, expiresAt: expiresAt}); loaded {
		return false
	}
	return true
}

type internalNonceClaim struct {
	value     string
	expiresAt time.Time
}

func internalRequestClaimValue(signingText string) string {
	sum := sha256.Sum256([]byte(signingText))
	return hex.EncodeToString(sum[:])
}