feat: 提供养成服务内部 IM 能力

server/internal_api.go

@@ -26,6 +26,7 @@ import (
 
 const (
 	internalCapabilityAdminQuery = "admin_query"
+	internalCapabilityCultivate  = "cultivate"
 
 	internalHeaderService   = "X-Service-Name"
 	internalHeaderTimestamp = "X-Timestamp"

server/internal_cultivate.go

@@ -0,0 +1,275 @@
+package main
+
+import (
+	"errors"
+	"net/http"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/tinode/chat/server/moderation"
+	"github.com/tinode/chat/server/push"
+	"github.com/tinode/chat/server/store"
+	"github.com/tinode/chat/server/store/types"
+)
+
+type cultivateUsersRequest struct {
+	UserIDs []int64 `json:"user_ids"`
+}
+
+type cultivateRelationsRequest struct {
+	ViewerID int64   `json:"viewer_id"`
+	PeerIDs  []int64 `json:"peer_ids"`
+}
+
+type cultivateFilesRequest struct {
+	UserID  int64    `json:"user_id"`
+	FileIDs []string `json:"file_ids"`
+}
+
+type cultivateReviewRequest struct {
+	ObjectType string `json:"object_type"`
+	ObjectID   int64  `json:"object_id"`
+	ActorID    int64  `json:"actor_id"`
+	Text       string `json:"text"`
+	TraceID    string `json:"trace_id"`
+}
+
+type cultivateEventRequest struct {
+	EventID      string         `json:"event_id"`
+	EventType    string         `json:"event_type"`
+	ActorID      int64          `json:"actor_id"`
+	TargetUserID int64          `json:"target_user_id"`
+	Title        string         `json:"title"`
+	Summary      string         `json:"summary"`
+	Action       map[string]any `json:"action"`
+	Payload      map[string]any `json:"payload"`
+}
+
+func serveInternalCultivate(w http.ResponseWriter, r *http.Request, prefix string) {
+	if r.Method != http.MethodPost {
+		writeIMError(w, r, http.StatusMethodNotAllowed, 30002, imReasonMethodNotAllowed, "method not allowed", nil)
+		return
+	}
+	if !authorizeInternalAPI(w, r, internalCapabilityCultivate) {
+		return
+	}
+	parts := splitInternalPath(strings.Trim(strings.TrimPrefix(r.URL.Path, prefix), "/"))
+	if len(parts) != 4 || parts[0] != "tenants" {
+		writeIMError(w, r, http.StatusNotFound, 30004, imReasonMalformedRequest, "resource not found", nil)
+		return
+	}
+	tenant, ok := resolveCultivateTenant(w, r, parts[1])
+	if !ok {
+		return
+	}
+	switch parts[2] + "/" + parts[3] {
+	case "users/batch":
+		serveCultivateUsers(w, r, tenant.ID)
+	case "relations/check":
+		serveCultivateRelations(w, r, tenant.ID)
+	case "files/validate":
+		serveCultivateFiles(w, r, tenant.ID)
+	case "moderation/review":
+		serveCultivateReview(w, r, tenant)
+	case "events/publish":
+		serveCultivateEvent(w, r, tenant.ID)
+	default:
+		writeIMError(w, r, http.StatusNotFound, 30004, imReasonMalformedRequest, "resource not found", nil)
+	}
+}
+
+func resolveCultivateTenant(w http.ResponseWriter, r *http.Request, value string) (*types.Tenant, bool) {
+	if id, err := strconv.ParseInt(value, 10, 64); err == nil && id > 0 && store.Tenants != nil {
+		tenant, lookupErr := store.Tenants.GetByID(types.TenantID(id))
+		if lookupErr == nil && tenant != nil && tenant.IsActive() {
+			return tenant, true
+		}
+		writeIMError(w, r, http.StatusNotFound, 30010, imReasonTenantUnavailable, "tenant unavailable", nil)
+		return nil, false
+	}
+	return resolveInternalTenant(w, r, value)
+}
+
+func serveCultivateUsers(w http.ResponseWriter, r *http.Request, tenantID types.TenantID) {
+	var body cultivateUsersRequest
+	if !decodeIMJSONBody(w, r, &body) {
+		return
+	}
+	if len(body.UserIDs) == 0 || len(body.UserIDs) > 200 {
+		writeIMError(w, r, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", nil)
+		return
+	}
+	items := make([]map[string]any, 0, len(body.UserIDs))
+	for _, rawID := range body.UserIDs {
+		if rawID <= 0 {
+			continue
+		}
+		uid := store.EncodeUid(rawID)
+		user, err := store.Users.Get(tenantID, uid)
+		if err != nil || user == nil {
+			continue
+		}
+		item := map[string]any{"user_id": rawID, "state": user.State.String(), "nickname": "", "avatar_file_id": "", "youth_mode": false}
+		if profile, err := store.Profiles.Get(tenantID, uid); err == nil && profile != nil {
+			item["nickname"], item["avatar_file_id"] = profile.Nickname, profile.AvatarFileID
+		}
+		if settings, err := store.Security.GetYouthMode(tenantID, uid); err == nil && settings != nil {
+			item["youth_mode"] = settings.Enabled
+		}
+		items = append(items, item)
+	}
+	writeIMSuccess(w, r, map[string]any{"items": items})
+}
+
+func serveCultivateRelations(w http.ResponseWriter, r *http.Request, tenantID types.TenantID) {
+	var body cultivateRelationsRequest
+	if !decodeIMJSONBody(w, r, &body) {
+		return
+	}
+	if body.ViewerID <= 0 || len(body.PeerIDs) > 500 {
+		writeIMError(w, r, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", nil)
+		return
+	}
+	viewer := store.EncodeUid(body.ViewerID)
+	items := make([]map[string]any, 0, len(body.PeerIDs))
+	for _, rawID := range body.PeerIDs {
+		if rawID <= 0 || rawID == body.ViewerID {
+			continue
+		}
+		peer := store.EncodeUid(rawID)
+		ctx, err := store.Relations.GetRelationshipContext(tenantID, viewer, peer, viewer.P2PName(peer))
+		if err != nil || ctx == nil {
+			continue
+		}
+		items = append(items, map[string]any{"peer_id": rawID, "relationship": ctx.Relationship, "can_send": ctx.CanSend})
+	}
+	youth := false
+	if settings, err := store.Security.GetYouthMode(tenantID, viewer); err == nil && settings != nil {
+		youth = settings.Enabled
+	}
+	writeIMSuccess(w, r, map[string]any{"items": items, "viewer_youth_mode": youth})
+}
+
+func serveCultivateFiles(w http.ResponseWriter, r *http.Request, tenantID types.TenantID) {
+	var body cultivateFilesRequest
+	if !decodeIMJSONBody(w, r, &body) {
+		return
+	}
+	if body.UserID <= 0 || len(body.FileIDs) == 0 || len(body.FileIDs) > 20 {
+		writeIMError(w, r, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", nil)
+		return
+	}
+	uid := store.EncodeUid(body.UserID)
+	items := make([]map[string]any, 0, len(body.FileIDs))
+	for _, supplied := range body.FileIDs {
+		fileID := normalizeIMUploadedFileID(supplied)
+		file, err := store.Files.Get(tenantID, fileID)
+		owner := types.ParseUid(fileOwner(file))
+		if owner.IsZero() {
+			owner = types.ParseUserId(fileOwner(file))
+		}
+		if err != nil || file == nil || file.Status != types.UploadCompleted || owner != uid {
+			writeIMError(w, r, http.StatusUnprocessableEntity, 30000, imReasonMalformedRequest, "invalid file", map[string]any{"file_id": supplied})
+			return
+		}
+		items = append(items, map[string]any{"file_id": fileID, "mime_type": file.MimeType, "size": file.Size, "review_status": cultivateFileReviewStatus(tenantID, fileID)})
+	}
+	writeIMSuccess(w, r, map[string]any{"items": items})
+}
+
+func fileOwner(file *types.FileDef) string {
+	if file == nil {
+		return ""
+	}
+	return file.User
+}
+
+func cultivateFileReviewStatus(tenantID types.TenantID, fileID string) string {
+	uid := types.ParseUid(fileID)
+	if uid.IsZero() {
+		return "unconfigured"
+	}
+	row, err := store.InternalAdminQueries.QueryOne(tenantID, `SELECT COUNT(*) AS total,
+		SUM(status='risky') AS risky_count,
+		SUM(status IN ('pending','running','failed')) AS pending_count
+		FROM im_file_review_tasks WHERE file_id=?`, []any{store.DecodeUid(uid)})
+	if err != nil {
+		return "unconfigured"
+	}
+	total, _ := numericID(row["total"])
+	risky, _ := numericID(row["risky_count"])
+	pending, _ := numericID(row["pending_count"])
+	if risky > 0 {
+		return "risky"
+	}
+	if pending > 0 {
+		return "pending"
+	}
+	if total > 0 {
+		return "clean"
+	}
+	return "unconfigured"
+}
+
+func serveCultivateReview(w http.ResponseWriter, r *http.Request, tenant *types.Tenant) {
+	var body cultivateReviewRequest
+	if !decodeIMJSONBody(w, r, &body) {
+		return
+	}
+	if body.ActorID <= 0 || body.ObjectID <= 0 {
+		writeIMError(w, r, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", nil)
+		return
+	}
+	scene := moderation.SceneFeedPost
+	if body.ObjectType == "comment" {
+		scene = moderation.SceneFeedComment
+	} else if body.ObjectType != "post" {
+		writeIMError(w, r, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", map[string]any{"field": "object_type"})
+		return
+	}
+	result, err := moderation.Review(r.Context(), moderation.Request{TenantID: tenant.ID, TenantCode: tenant.Code, Scene: scene, ActorID: store.EncodeUid(body.ActorID), Text: body.Text, TraceID: body.TraceID})
+	if result == nil {
+		result = &moderation.Result{Action: moderation.ActionAuditOnly}
+	}
+	data := map[string]any{"request_id": result.RequestID, "action": result.Action, "risk_level": result.RiskLevel, "risk_labels": result.RiskLabels, "provider": result.Provider, "provider_request_id": result.ProviderRequestID}
+	if errors.Is(err, moderation.ErrUnavailable) {
+		writeIMError(w, r, http.StatusServiceUnavailable, 50001, imReasonInternal, "content review unavailable", data)
+		return
+	}
+	// A moderation rejection is a valid review decision, not a transport error.
+	writeIMSuccess(w, r, data)
+}
+
+func serveCultivateEvent(w http.ResponseWriter, r *http.Request, tenantID types.TenantID) {
+	var body cultivateEventRequest
+	if !decodeIMJSONBody(w, r, &body) {
+		return
+	}
+	if body.EventID == "" || body.EventType == "" || body.TargetUserID <= 0 {
+		writeIMError(w, r, http.StatusBadRequest, 30000, imReasonMalformedRequest, "malformed request", nil)
+		return
+	}
+	target, actor := store.EncodeUid(body.TargetUserID), store.EncodeUid(body.ActorID)
+	if body.Title != "" {
+		if err := createIMSystemNotification(tenantID, target, stableIMEventID(body.EventID, "notification"), body.EventType, body.Title, body.Summary, types.KVMap(body.Action)); err != nil {
+			if errors.Is(err, types.ErrDuplicate) {
+				writeIMSuccess(w, r, map[string]any{"event_id": body.EventID, "duplicate": true})
+				return
+			}
+			writeIMError(w, r, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil)
+			return
+		}
+	}
+	if err := enqueueIMBusinessEvent(tenantID, body.EventID, "user", target.UserId(), actor, body.EventType, body.Payload); err != nil {
+		writeIMError(w, r, http.StatusInternalServerError, 90000, imReasonInternal, "internal error", nil)
+		return
+	}
+	if body.Title != "" {
+		sendPush(&push.Receipt{TenantID: tenantID, To: map[types.Uid]push.Recipient{target: {}}, Payload: push.Payload{
+			What: push.ActMsg, Topic: "sys", Timestamp: time.Now().UTC(), ContentType: "application/json",
+			Content: map[string]any{"event_id": body.EventID, "event_type": body.EventType, "title": body.Title, "summary": body.Summary, "action": body.Action},
+		}})
+	}
+	writeIMSuccess(w, r, map[string]any{"event_id": body.EventID})
+}

server/internal_cultivate_test.go

@@ -0,0 +1,36 @@
+package main
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+)
+
+func TestInternalCultivateRequiresSignedServiceRequest(t *testing.T) {
+	previous := internalAPI
+	defer func() { internalAPI = previous }()
+	internalAPI = internalAPIRuntime{config: internalAPIConfig{Enabled: true, Services: map[string]internalAPIServiceConfig{"cultivate-server": {Secret: "secret", Allow: []string{internalCapabilityCultivate}}}}}
+	req := httptest.NewRequest(http.MethodPost, "/internal/cultivate/tenants/default/users/batch", strings.NewReader(`{"user_ids":[1]}`))
+	rec := httptest.NewRecorder()
+	serveInternalCultivate(rec, req, "/internal/cultivate/")
+	if rec.Code != http.StatusUnauthorized {
+		t.Fatalf("status=%d body=%s", rec.Code, rec.Body.String())
+	}
+}
+
+func TestInternalCultivateRejectsWrongCapability(t *testing.T) {
+	previous := internalAPI
+	defer func() { internalAPI = previous }()
+	internalAPI = internalAPIRuntime{config: internalAPIConfig{Enabled: true, Services: map[string]internalAPIServiceConfig{"cultivate-server": {Secret: "secret", Allow: []string{internalCapabilityAdminQuery}}}}}
+	req := httptest.NewRequest(http.MethodPost, "/internal/cultivate/tenants/default/users/batch", strings.NewReader(`{"user_ids":[1]}`))
+	req.Header.Set(internalHeaderService, "cultivate-server")
+	req.Header.Set(internalHeaderTimestamp, "1")
+	req.Header.Set(internalHeaderNonce, "nonce")
+	req.Header.Set(internalHeaderSignature, "signature")
+	rec := httptest.NewRecorder()
+	serveInternalCultivate(rec, req, "/internal/cultivate/")
+	if rec.Code != http.StatusForbidden {
+		t.Fatalf("status=%d body=%s", rec.Code, rec.Body.String())
+	}
+}

server/main.go

@@ -928,6 +928,10 @@ func main() {
 	mux.HandleFunc(adminQueryPrefix, func(wrt http.ResponseWriter, req *http.Request) {
 		serveInternalAdminQuery(wrt, req, adminQueryPrefix)
 	})
+	cultivatePrefix := config.ApiPath + "internal/cultivate/"
+	mux.HandleFunc(cultivatePrefix, func(wrt http.ResponseWriter, req *http.Request) {
+		serveInternalCultivate(wrt, req, cultivatePrefix)
+	})
 	// Handle long polling clients. Enable compression.
 	mux.Handle(config.ApiPath+"v0/channels/lp", gh.CompressHandler(http.HandlerFunc(serveLongPoll)))
 	if config.Media != nil {

server/meetingim.conf

@@ -150,6 +150,10 @@
       "hotgo-admin": {
         "secret": "8569f5626a54950539eece9f731c9e3bf073daa74af8d23735d19afacac40f4b",
         "allow": ["admin_query"]
+      },
+      "cultivate-server": {
+        "secret": "cultivate-local-internal-secret-change-me",
+        "allow": ["cultivate"]
       }
     }
   },

server/moderation/moderation_test.go

@@ -2,6 +2,8 @@ package moderation
 
 import (
 	"context"
+	"sort"
+	"strings"
 	"testing"
 	"time"
 
@@ -33,7 +35,12 @@ func (testAudit) RecordContentModeration(types.TenantID, *types.ContentModeratio
 
 func TestExtractText(t *testing.T) {
 	content := map[string]any{"txt": "正文", "ent": []any{map[string]any{"data": map[string]any{"title": "查看详情", "url": "https://example.com"}}}, "meta": "ignore"}
-	if got := ExtractText(content); got != "正文\n查看详情" && got != "查看详情\n正文" {
+	gotParts := strings.Split(ExtractText(content), "\n")
+	sort.Strings(gotParts)
+	wantParts := []string{"https://example.com", "查看详情", "正文"}
+	sort.Strings(wantParts)
+	if strings.Join(gotParts, "\n") != strings.Join(wantParts, "\n") {
+		got := ExtractText(content)
 		t.Fatalf("ExtractText()=%q", got)
 	}
 }

server/moderation/provider.go

@@ -16,6 +16,8 @@ const (
 	ScenePrivateMessage    = "private_message"
 	SceneGroupMessage      = "group_message"
 	SceneGroupAnnouncement = "group_announcement"
+	SceneFeedPost          = "feed_post"
+	SceneFeedComment       = "feed_comment"
 
 	ActionAllow     = "allow"
 	ActionBlock     = "block"

server/tinode.conf

@@ -150,6 +150,10 @@
       "hotgo-admin": {
         "secret": "8569f5626a54950539eece9f731c9e3bf073daa74af8d23735d19afacac40f4b",
         "allow": ["admin_query"]
+      },
+      "cultivate-server": {
+        "secret": "cultivate-local-internal-secret-change-me",
+        "allow": ["cultivate"]
       }
     }
   },