package main import ( "context" "encoding/json" "time" "github.com/tinode/chat/server/store/types" ) type kafkaWorkbenchPayload struct { Module string `json:"module"` Action string `json:"action"` ResourceID string `json:"resource_id"` ActorUserID string `json:"actor_user_id"` TargetUserIDs []string `json:"target_user_ids,omitempty"` Title string `json:"title,omitempty"` Summary string `json:"summary,omitempty"` Data map[string]any `json:"data,omitempty"` OccurredAt time.Time `json:"occurred_at"` } type kafkaAuditPayload struct { ActorType string `json:"actor_type"` ActorID string `json:"actor_id"` Action string `json:"action"` Resource string `json:"resource"` ResourceID string `json:"resource_id"` Result string `json:"result"` Reason string `json:"reason,omitempty"` Data map[string]any `json:"data,omitempty"` OccurredAt time.Time `json:"occurred_at"` } var publishKafkaAuditEventEnvelope = publishKafkaEnvelope func enqueueKafkaWorkbenchEvent(ctx context.Context, tenantID types.TenantID, actor types.Uid, module, action, resourceID string, targets []types.Uid, title, summary string, data map[string]any) error { targetUserIDs := make([]string, 0, len(targets)) for _, target := range targets { if !target.IsZero() { targetUserIDs = append(targetUserIDs, target.UserId()) } } payload := kafkaWorkbenchPayload{ Module: module, Action: action, ResourceID: resourceID, ActorUserID: actor.UserId(), TargetUserIDs: targetUserIDs, Title: title, Summary: summary, Data: data, OccurredAt: time.Now().UTC().Round(time.Millisecond), } envelope := kafkaEnvelope{ EventID: stableIMEventID("workbench-sync", tenantID, "workbench."+action, resourceID, payload.OccurredAt), TenantID: tenantID, EventType: "workbench." + action, BizType: "workbench", BizID: resourceID, } return executeKafkaWorkbenchEvent(tenantID, envelope, payload) } func enqueueKafkaAuditEvent(ctx context.Context, tenantID types.TenantID, actorType string, actor types.Uid, action, resource, resourceID, result, reason string, data map[string]any) error { if !kafkaEnabled() { return nil } actorID := "" if !actor.IsZero() { actorID = actor.UserId() } payload := kafkaAuditPayload{ ActorType: actorType, ActorID: actorID, Action: action, Resource: resource, ResourceID: resourceID, Result: result, Reason: reason, Data: data, OccurredAt: time.Now().UTC().Round(time.Millisecond), } encoded, err := json.Marshal(payload) if err != nil { return err } envelope := kafkaEnvelope{ EventID: stableIMEventID("audit_event", tenantID, "audit."+action, resource+":"+resourceID, encoded), TenantID: tenantID, SourceService: "im-server", TargetService: "hotgo-admin", EventType: "audit." + action, BizType: "audit", BizID: resource + ":" + resourceID, OccurredAt: payload.OccurredAt, Payload: encoded, } if !actor.IsZero() { envelope.Actor = &kafkaActor{Type: actorType, ID: actorID} } return publishKafkaAuditEventEnvelope(ctx, "audit_event", envelope.BizID, envelope) } func executeKafkaWorkbenchEvent(tenantID types.TenantID, envelope kafkaEnvelope, payload kafkaWorkbenchPayload) error { actor := types.ParseUserId(payload.ActorUserID) if actor.IsZero() { actor = types.ParseUid(payload.ActorUserID) } for _, raw := range payload.TargetUserIDs { target := types.ParseUserId(raw) if target.IsZero() { continue } eventID := stableIMEventID("workbench", envelope.EventID, target.UserId()) if target != actor && payload.Title != "" { if err := createIMSystemNotification(tenantID, target, stableIMEventID(eventID, "notification"), payload.Action, payload.Title, payload.Summary, types.KVMap{"module": payload.Module, "resource_id": payload.ResourceID}); err != nil { return err } } if err := enqueueIMBusinessEvent(tenantID, eventID, "user", target.UserId(), actor, payload.Action, payload.Data); err != nil { return err } } return nil }