130 lines
4.2 KiB
Go
130 lines
4.2 KiB
Go
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
|
|
}
|