mirror of
https://github.com/tiennm99/goclaw.git
synced 2026-10-12 22:14:37 +00:00
* feat(channelmemory): add passive extraction tuning * feat(ui): expose passive memory tuning controls * docs(memory): document passive extraction tuning --------- Co-authored-by: Collective Developer <man@collective.dev>
807 lines
31 KiB
Go
807 lines
31 KiB
Go
package http
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"log/slog"
|
|
"net/http"
|
|
"strconv"
|
|
"strings"
|
|
|
|
"github.com/google/uuid"
|
|
|
|
"github.com/nextlevelbuilder/goclaw/internal/bus"
|
|
"github.com/nextlevelbuilder/goclaw/internal/channelmemory"
|
|
"github.com/nextlevelbuilder/goclaw/internal/channels"
|
|
"github.com/nextlevelbuilder/goclaw/internal/config"
|
|
"github.com/nextlevelbuilder/goclaw/internal/i18n"
|
|
"github.com/nextlevelbuilder/goclaw/internal/permissions"
|
|
"github.com/nextlevelbuilder/goclaw/internal/store"
|
|
"github.com/nextlevelbuilder/goclaw/pkg/protocol"
|
|
)
|
|
|
|
// OrphanChannelCleaner runs channel-type-specific cleanup when a delete
|
|
// arrives for a channel that's NOT loaded in the runtime Manager (e.g. it
|
|
// was disabled so InstanceLoader removed it). Closure injected from
|
|
// cmd/gateway.go captures the per-channel dependencies (portal store,
|
|
// encryption key) so the handler doesn't need to import per-channel
|
|
// packages directly.
|
|
//
|
|
// Returns nil for no-op cases (nothing to clean) so callers can ignore the
|
|
// error in those situations; real failures (store read, decode) propagate.
|
|
type OrphanChannelCleaner func(ctx context.Context, tenantID uuid.UUID, configJSON []byte) error
|
|
|
|
// ChannelInstancesHandler handles channel instance CRUD endpoints.
|
|
type ChannelInstancesHandler struct {
|
|
store store.ChannelInstanceStore
|
|
agentStore store.AgentStore
|
|
configPermStore store.ConfigPermissionStore
|
|
contactStore store.ContactStore
|
|
tenantStore store.TenantStore
|
|
msgBus *bus.MessageBus
|
|
memberResolver channels.MemberResolver // optional — enriches file_writer metadata on addwriter
|
|
channelMgr *channels.Manager // optional — enables ChannelDestroyer hook on delete
|
|
memoryService *channelmemory.Service
|
|
mcpStore store.MCPServerStore
|
|
secureCLIStore store.SecureCLIStore
|
|
mcpContextStore store.MCPContextAdminStore
|
|
cliContextStore store.SecureCLIContextAdminStore
|
|
// orphanCleaners is keyed by channel_type; called when channelMgr.GetChannel
|
|
// returns false. Keeps handler agnostic of per-channel packages.
|
|
orphanCleaners map[string]OrphanChannelCleaner
|
|
}
|
|
|
|
// NewChannelInstancesHandler creates a handler for channel instance management endpoints.
|
|
func NewChannelInstancesHandler(s store.ChannelInstanceStore, agentStore store.AgentStore, configPermStore store.ConfigPermissionStore, contactStore store.ContactStore, tenantStore store.TenantStore, msgBus *bus.MessageBus) *ChannelInstancesHandler {
|
|
return &ChannelInstancesHandler{store: s, agentStore: agentStore, configPermStore: configPermStore, contactStore: contactStore, tenantStore: tenantStore, msgBus: msgBus}
|
|
}
|
|
|
|
// SetMemberResolver wires a channel member resolver so addwriter can auto-fill
|
|
// metadata when the caller supplies neither DisplayName nor Username.
|
|
func (h *ChannelInstancesHandler) SetMemberResolver(r channels.MemberResolver) {
|
|
h.memberResolver = r
|
|
}
|
|
|
|
// SetChannelManager wires the channel Manager so handleDelete can invoke
|
|
// ChannelDestroyer.Destroy() before removing the DB row — required for
|
|
// Bitrix24 to imbot.unregister its bot on portal-side. Setter-pattern (vs
|
|
// constructor param) because the Manager is created AFTER this handler
|
|
// in cmd/gateway.go's startup ordering.
|
|
func (h *ChannelInstancesHandler) SetChannelManager(mgr *channels.Manager) {
|
|
h.channelMgr = mgr
|
|
if h.memoryService != nil {
|
|
h.memoryService.ContextResolver = channelmemory.ContextResolverFunc(func(ctx context.Context, inst *store.ChannelInstanceData, group store.PendingMessageGroup) (channelmemory.ExtractionContext, error) {
|
|
return resolveMemoryExtractionContextFromManager(ctx, mgr, inst, group)
|
|
})
|
|
}
|
|
}
|
|
|
|
type memoryExtractionContextProvider interface {
|
|
ResolveMemoryExtractionContext(ctx context.Context, inst *store.ChannelInstanceData, group store.PendingMessageGroup) (channelmemory.ExtractionContext, error)
|
|
}
|
|
|
|
func resolveMemoryExtractionContextFromManager(ctx context.Context, mgr *channels.Manager, inst *store.ChannelInstanceData, group store.PendingMessageGroup) (channelmemory.ExtractionContext, error) {
|
|
if mgr == nil || inst == nil || inst.Name == "" {
|
|
return channelmemory.ExtractionContext{}, nil
|
|
}
|
|
ch, ok := mgr.GetChannel(inst.Name)
|
|
if !ok {
|
|
return channelmemory.ExtractionContext{}, nil
|
|
}
|
|
provider, ok := ch.(memoryExtractionContextProvider)
|
|
if !ok {
|
|
return channelmemory.ExtractionContext{}, nil
|
|
}
|
|
return provider.ResolveMemoryExtractionContext(ctx, inst, group)
|
|
}
|
|
|
|
// SetCapabilityStores wires MCP and Secure CLI stores for channel-context
|
|
// capability visibility. Kept as a setter to preserve startup ordering.
|
|
func (h *ChannelInstancesHandler) SetCapabilityStores(mcpStore store.MCPServerStore, secureCLIStore store.SecureCLIStore) {
|
|
h.mcpStore = mcpStore
|
|
h.secureCLIStore = secureCLIStore
|
|
if contextStore, ok := mcpStore.(store.MCPContextAdminStore); ok {
|
|
h.mcpContextStore = contextStore
|
|
}
|
|
if contextStore, ok := secureCLIStore.(store.SecureCLIContextAdminStore); ok {
|
|
h.cliContextStore = contextStore
|
|
}
|
|
}
|
|
|
|
// RegisterOrphanCleaner registers a per-channel-type cleanup function that
|
|
// fires during handleDelete when the channel is no longer loaded in the
|
|
// Manager (typically because admin disabled it). Without this, deleting a
|
|
// disabled Bitrix24 channel leaves the bot as a zombie on the portal.
|
|
func (h *ChannelInstancesHandler) RegisterOrphanCleaner(channelType string, fn OrphanChannelCleaner) {
|
|
if h.orphanCleaners == nil {
|
|
h.orphanCleaners = make(map[string]OrphanChannelCleaner)
|
|
}
|
|
h.orphanCleaners[channelType] = fn
|
|
}
|
|
|
|
// RegisterRoutes registers all channel instance routes on the given mux.
|
|
func (h *ChannelInstancesHandler) RegisterRoutes(mux *http.ServeMux) {
|
|
// Channel instance CRUD (reads: viewer+, writes: admin+)
|
|
mux.HandleFunc("GET /v1/channels/instances", h.auth(h.handleList))
|
|
mux.HandleFunc("POST /v1/channels/instances", h.adminAuth(h.handleCreate))
|
|
mux.HandleFunc("GET /v1/channels/instances/{id}", h.auth(h.handleGet))
|
|
mux.HandleFunc("PUT /v1/channels/instances/{id}", h.adminAuth(h.handleUpdate))
|
|
mux.HandleFunc("DELETE /v1/channels/instances/{id}", h.adminAuth(h.handleDelete))
|
|
mux.HandleFunc("POST /v1/channels/instances/{id}/metadata/refresh", h.adminAuth(h.handleRefreshChannelMetadata))
|
|
|
|
// Channel contacts (global, not per-agent)
|
|
if h.contactStore != nil {
|
|
mux.HandleFunc("GET /v1/contacts", h.auth(h.handleListContacts))
|
|
mux.HandleFunc("GET /v1/contacts/resolve", h.auth(h.handleResolveContacts))
|
|
mux.HandleFunc("POST /v1/contacts/merge", h.adminAuth(h.handleMergeContacts))
|
|
mux.HandleFunc("POST /v1/contacts/unmerge", h.adminAuth(h.handleUnmergeContacts))
|
|
mux.HandleFunc("GET /v1/contacts/merged/{tenantUserId}", h.auth(h.handleListMergedContacts))
|
|
}
|
|
if h.tenantStore != nil {
|
|
mux.HandleFunc("GET /v1/tenant-users", h.auth(h.handleListTenantUsers))
|
|
}
|
|
|
|
// Unified user search (contacts + tenant_users)
|
|
if h.contactStore != nil {
|
|
mux.HandleFunc("GET /v1/users/search", h.auth(h.handleSearchUsers))
|
|
}
|
|
|
|
// Group file writers (nested under channel instances)
|
|
if h.configPermStore != nil {
|
|
mux.HandleFunc("GET /v1/channels/instances/{id}/writers/groups", h.auth(h.handleWriterGroups))
|
|
mux.HandleFunc("GET /v1/channels/instances/{id}/writers", h.auth(h.handleListWriters))
|
|
mux.HandleFunc("POST /v1/channels/instances/{id}/writers/test", h.auth(h.handleTestWriter))
|
|
mux.HandleFunc("POST /v1/channels/instances/{id}/writers", h.adminAuth(h.handleAddWriter))
|
|
mux.HandleFunc("DELETE /v1/channels/instances/{id}/writers/{userId}", h.adminAuth(h.handleRemoveWriter))
|
|
}
|
|
|
|
if h.contactStore != nil {
|
|
mux.HandleFunc("GET /v1/channels/instances/{id}/contexts", h.auth(h.handleListContexts))
|
|
mux.HandleFunc("GET /v1/channels/instances/{id}/contexts/{scopeType}/{scopeKey}/members", h.auth(h.handleListContextMembers))
|
|
}
|
|
mux.HandleFunc("GET /v1/channels/instances/{id}/contexts/{scopeType}/{scopeKey}/capabilities", h.auth(h.handleListContextCapabilities))
|
|
mux.HandleFunc("GET /v1/channels/instances/{id}/contexts/{scopeType}/{scopeKey}/mcp-grants", h.adminAuth(h.handleListContextMCPGrants))
|
|
mux.HandleFunc("PUT /v1/channels/instances/{id}/contexts/{scopeType}/{scopeKey}/mcp-grants/{serverID}", h.adminAuth(h.handleUpsertContextMCPGrant))
|
|
mux.HandleFunc("DELETE /v1/channels/instances/{id}/contexts/{scopeType}/{scopeKey}/mcp-grants/{serverID}", h.adminAuth(h.handleDeleteContextMCPGrant))
|
|
mux.HandleFunc("GET /v1/channels/instances/{id}/contexts/{scopeType}/{scopeKey}/mcp-credentials", h.adminAuth(h.handleListContextMCPCredentials))
|
|
mux.HandleFunc("PUT /v1/channels/instances/{id}/contexts/{scopeType}/{scopeKey}/mcp-credentials/{serverID}", h.adminAuth(h.handleSetContextMCPCredentials))
|
|
mux.HandleFunc("DELETE /v1/channels/instances/{id}/contexts/{scopeType}/{scopeKey}/mcp-credentials/{serverID}", h.adminAuth(h.handleDeleteContextMCPCredentials))
|
|
mux.HandleFunc("GET /v1/channels/instances/{id}/contexts/{scopeType}/{scopeKey}/cli-grants", h.adminAuth(h.handleListContextCLIGrants))
|
|
mux.HandleFunc("PUT /v1/channels/instances/{id}/contexts/{scopeType}/{scopeKey}/cli-grants/{binaryID}", h.adminAuth(h.handleUpsertContextCLIGrant))
|
|
mux.HandleFunc("DELETE /v1/channels/instances/{id}/contexts/{scopeType}/{scopeKey}/cli-grants/{binaryID}", h.adminAuth(h.handleDeleteContextCLIGrant))
|
|
mux.HandleFunc("GET /v1/channels/instances/{id}/contexts/{scopeType}/{scopeKey}/cli-credentials", h.adminAuth(h.handleListContextCLICredentials))
|
|
mux.HandleFunc("PUT /v1/channels/instances/{id}/contexts/{scopeType}/{scopeKey}/cli-credentials/{binaryID}", h.adminAuth(h.handleSetContextCLICredentials))
|
|
mux.HandleFunc("DELETE /v1/channels/instances/{id}/contexts/{scopeType}/{scopeKey}/cli-credentials/{binaryID}", h.adminAuth(h.handleDeleteContextCLICredentials))
|
|
|
|
if h.memoryService != nil {
|
|
mux.HandleFunc("GET /v1/channels/instances/{id}/memory-extraction", h.auth(h.handleMemoryExtractionStatus))
|
|
mux.HandleFunc("PUT /v1/channels/instances/{id}/memory-extraction/settings", h.adminAuth(h.handleMemoryExtractionSettings))
|
|
mux.HandleFunc("POST /v1/channels/instances/{id}/memory-extraction/run", h.adminAuth(h.handleMemoryExtractionRun))
|
|
mux.HandleFunc("POST /v1/channels/instances/{id}/memory-extraction/run-all", h.adminAuth(h.handleMemoryExtractionRunAll))
|
|
mux.HandleFunc("GET /v1/channels/instances/{id}/memory-extraction/groups", h.auth(h.handleMemoryExtractionGroups))
|
|
mux.HandleFunc("GET /v1/channels/instances/{id}/memory-extraction/items", h.auth(h.handleMemoryExtractionItems))
|
|
mux.HandleFunc("POST /v1/channels/instances/{id}/memory-extraction/items/{itemID}/approve", h.adminAuth(h.handleMemoryExtractionApprove))
|
|
mux.HandleFunc("POST /v1/channels/instances/{id}/memory-extraction/items/{itemID}/reject", h.adminAuth(h.handleMemoryExtractionReject))
|
|
mux.HandleFunc("DELETE /v1/channels/instances/{id}/memory-extraction/items/{itemID}", h.adminAuth(h.handleMemoryExtractionDelete))
|
|
}
|
|
}
|
|
|
|
func (h *ChannelInstancesHandler) SetMemoryExtractionService(svc *channelmemory.Service) {
|
|
h.memoryService = svc
|
|
}
|
|
|
|
func (h *ChannelInstancesHandler) auth(next http.HandlerFunc) http.HandlerFunc {
|
|
return requireAuth("", next)
|
|
}
|
|
|
|
func (h *ChannelInstancesHandler) adminAuth(next http.HandlerFunc) http.HandlerFunc {
|
|
return requireAuth(permissions.RoleAdmin, next)
|
|
}
|
|
|
|
func (h *ChannelInstancesHandler) emitCacheInvalidate(key string) {
|
|
if h.msgBus == nil {
|
|
return
|
|
}
|
|
h.msgBus.Broadcast(bus.Event{
|
|
Name: protocol.EventCacheInvalidate,
|
|
Payload: bus.CacheInvalidatePayload{Kind: bus.CacheKindChannelInstances, Key: key},
|
|
})
|
|
}
|
|
|
|
func (h *ChannelInstancesHandler) handleList(w http.ResponseWriter, r *http.Request) {
|
|
opts := store.ChannelInstanceListOpts{
|
|
Limit: 50,
|
|
Offset: 0,
|
|
}
|
|
|
|
if v := r.URL.Query().Get("search"); v != "" {
|
|
opts.Search = v
|
|
}
|
|
if v := r.URL.Query().Get("limit"); v != "" {
|
|
if n, err := strconv.Atoi(v); err == nil && n > 0 && n <= 200 {
|
|
opts.Limit = n
|
|
}
|
|
}
|
|
if v := r.URL.Query().Get("offset"); v != "" {
|
|
if n, err := strconv.Atoi(v); err == nil && n >= 0 {
|
|
opts.Offset = n
|
|
}
|
|
}
|
|
|
|
instances, err := h.store.ListPaged(r.Context(), opts)
|
|
if err != nil {
|
|
slog.Error("channel_instances.list", "error", err)
|
|
locale := store.LocaleFromContext(r.Context())
|
|
writeError(w, http.StatusInternalServerError, protocol.ErrInternal, i18n.T(locale, i18n.MsgFailedToList, "instances"))
|
|
return
|
|
}
|
|
|
|
total, _ := h.store.CountInstances(r.Context(), opts)
|
|
|
|
result := make([]map[string]any, 0, len(instances))
|
|
for _, inst := range instances {
|
|
result = append(result, maskInstanceHTTP(inst))
|
|
}
|
|
|
|
writeJSON(w, http.StatusOK, map[string]any{
|
|
"instances": result,
|
|
"total": total,
|
|
"limit": opts.Limit,
|
|
"offset": opts.Offset,
|
|
})
|
|
}
|
|
|
|
func (h *ChannelInstancesHandler) handleCreate(w http.ResponseWriter, r *http.Request) {
|
|
locale := store.LocaleFromContext(r.Context())
|
|
var body struct {
|
|
Name string `json:"name"`
|
|
DisplayName string `json:"display_name"`
|
|
ChannelType string `json:"channel_type"`
|
|
AgentID string `json:"agent_id"`
|
|
Credentials json.RawMessage `json:"credentials"`
|
|
Config json.RawMessage `json:"config"`
|
|
Enabled *bool `json:"enabled"`
|
|
}
|
|
if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20)).Decode(&body); err != nil {
|
|
writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidJSON))
|
|
return
|
|
}
|
|
|
|
if body.Name == "" || body.ChannelType == "" || body.AgentID == "" {
|
|
writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgRequired, "name, channel_type, and agent_id"))
|
|
return
|
|
}
|
|
|
|
if !isValidChannelType(body.ChannelType) {
|
|
writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidChannelType))
|
|
return
|
|
}
|
|
|
|
agentID, err := uuid.Parse(body.AgentID)
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidID, "agent"))
|
|
return
|
|
}
|
|
|
|
enabled := true
|
|
if body.Enabled != nil {
|
|
enabled = *body.Enabled
|
|
}
|
|
|
|
userID := store.UserIDFromContext(r.Context())
|
|
|
|
inst := &store.ChannelInstanceData{
|
|
Name: body.Name,
|
|
DisplayName: body.DisplayName,
|
|
ChannelType: body.ChannelType,
|
|
AgentID: agentID,
|
|
Credentials: body.Credentials,
|
|
Config: config.NormalizeChannelInstanceConfigRaw(body.ChannelType, body.Config),
|
|
Enabled: enabled,
|
|
CreatedBy: userID,
|
|
}
|
|
|
|
if err := h.store.Create(r.Context(), inst); err != nil {
|
|
slog.Error("channel_instances.create", "error", err)
|
|
writeError(w, http.StatusInternalServerError, protocol.ErrInternal, i18n.T(locale, i18n.MsgFailedToCreate, "channel instance", "internal error"))
|
|
return
|
|
}
|
|
|
|
h.emitCacheInvalidate(inst.ID.String())
|
|
emitAudit(h.msgBus, r, "channel_instance.created", "channel_instance", inst.ID.String())
|
|
writeJSON(w, http.StatusCreated, maskInstanceHTTP(*inst))
|
|
}
|
|
|
|
func (h *ChannelInstancesHandler) handleGet(w http.ResponseWriter, r *http.Request) {
|
|
locale := store.LocaleFromContext(r.Context())
|
|
id, err := uuid.Parse(r.PathValue("id"))
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidID, "instance"))
|
|
return
|
|
}
|
|
|
|
inst, err := h.store.Get(r.Context(), id)
|
|
if err != nil {
|
|
writeError(w, http.StatusNotFound, protocol.ErrNotFound, i18n.T(locale, i18n.MsgInstanceNotFound))
|
|
return
|
|
}
|
|
|
|
writeJSON(w, http.StatusOK, maskInstanceHTTP(*inst))
|
|
}
|
|
|
|
func (h *ChannelInstancesHandler) handleUpdate(w http.ResponseWriter, r *http.Request) {
|
|
locale := store.LocaleFromContext(r.Context())
|
|
id, err := uuid.Parse(r.PathValue("id"))
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidID, "instance"))
|
|
return
|
|
}
|
|
|
|
var updates map[string]any
|
|
if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20)).Decode(&updates); err != nil {
|
|
writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidJSON))
|
|
return
|
|
}
|
|
|
|
// Allowlist: only permit known channel instance columns.
|
|
updates = filterAllowedKeys(updates, channelInstanceAllowedFields)
|
|
h.normalizeChannelInstanceConfigUpdate(r.Context(), id, updates)
|
|
|
|
if err := h.store.Update(r.Context(), id, updates); err != nil {
|
|
slog.Error("channel_instances.update", "error", err)
|
|
writeError(w, http.StatusInternalServerError, protocol.ErrInternal, i18n.T(locale, i18n.MsgFailedToUpdate, "channel instance", "internal error"))
|
|
return
|
|
}
|
|
|
|
h.emitCacheInvalidate("")
|
|
emitAudit(h.msgBus, r, "channel_instance.updated", "channel_instance", id.String())
|
|
writeJSON(w, http.StatusOK, map[string]string{"status": "updated"})
|
|
}
|
|
|
|
func (h *ChannelInstancesHandler) normalizeChannelInstanceConfigUpdate(ctx context.Context, id uuid.UUID, updates map[string]any) {
|
|
value, ok := updates["config"]
|
|
if !ok {
|
|
return
|
|
}
|
|
channelType, _ := updates["channel_type"].(string)
|
|
if channelType == "" {
|
|
if inst, err := h.store.Get(ctx, id); err == nil {
|
|
channelType = inst.ChannelType
|
|
}
|
|
}
|
|
updates["config"] = config.NormalizeChannelInstanceConfigValue(channelType, value)
|
|
}
|
|
|
|
func (h *ChannelInstancesHandler) handleDelete(w http.ResponseWriter, r *http.Request) {
|
|
locale := store.LocaleFromContext(r.Context())
|
|
id, err := uuid.Parse(r.PathValue("id"))
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidID, "instance"))
|
|
return
|
|
}
|
|
|
|
// Look up instance to check if it's a default (seeded) instance.
|
|
inst, err := h.store.Get(r.Context(), id)
|
|
if err != nil {
|
|
writeError(w, http.StatusNotFound, protocol.ErrNotFound, i18n.T(locale, i18n.MsgInstanceNotFound))
|
|
return
|
|
}
|
|
if store.IsDefaultChannelInstance(inst.Name) {
|
|
writeError(w, http.StatusForbidden, protocol.ErrUnauthorized, i18n.T(locale, i18n.MsgCannotDeleteDefaultInst))
|
|
return
|
|
}
|
|
|
|
// Best-effort: notify the channel impl so external resources (e.g. the
|
|
// Bitrix24 imbot.register'd bot) get cleaned up BEFORE the DB row is
|
|
// removed. Order matters: deleting the row first triggers a cache
|
|
// invalidate → InstanceLoader Reload Stop's the channel and clears
|
|
// in-memory botID, leaving the upstream bot orphaned.
|
|
//
|
|
// Two paths:
|
|
// 1. Channel still loaded in Manager → ChannelDestroyer.Destroy() —
|
|
// uses cached botID, calls imbot.unregister directly via the live
|
|
// Client. This is the normal path.
|
|
// 2. Channel NOT in Manager (e.g. admin disabled it earlier, so
|
|
// InstanceLoader.Reload removed it) → fall back to a registered
|
|
// orphan cleaner for this channel type. Reads bot_id from
|
|
// persisted portal state. Without this branch, deleting a disabled
|
|
// Bitrix24 channel orphans the bot on the portal.
|
|
//
|
|
// Channels without external state (Telegram, Discord, Slack, …) don't
|
|
// implement ChannelDestroyer AND don't register an orphan cleaner —
|
|
// both branches no-op for them.
|
|
if h.channelMgr != nil {
|
|
if ch, ok := h.channelMgr.GetChannel(inst.Name); ok {
|
|
if destroyer, ok := ch.(channels.ChannelDestroyer); ok {
|
|
if err := destroyer.Destroy(r.Context()); err != nil {
|
|
slog.Warn("channel_instances.delete: destroyer failed — proceeding with DB delete",
|
|
"name", inst.Name, "tenant_id", inst.TenantID, "type", inst.ChannelType, "err", err)
|
|
}
|
|
}
|
|
} else if cleaner, ok := h.orphanCleaners[inst.ChannelType]; ok && cleaner != nil {
|
|
if err := cleaner(r.Context(), inst.TenantID, inst.Config); err != nil {
|
|
slog.Warn("channel_instances.delete: orphan cleaner failed — proceeding with DB delete",
|
|
"name", inst.Name, "tenant_id", inst.TenantID, "type", inst.ChannelType, "err", err)
|
|
}
|
|
}
|
|
}
|
|
|
|
if err := h.store.Delete(r.Context(), id); err != nil {
|
|
slog.Error("channel_instances.delete", "error", err)
|
|
writeError(w, http.StatusInternalServerError, protocol.ErrInternal, i18n.T(locale, i18n.MsgFailedToDelete, "channel instance", "internal error"))
|
|
return
|
|
}
|
|
|
|
h.emitCacheInvalidate("")
|
|
emitAudit(h.msgBus, r, "channel_instance.deleted", "channel_instance", id.String())
|
|
writeJSON(w, http.StatusOK, map[string]string{"status": "deleted"})
|
|
}
|
|
|
|
// maskInstanceHTTP returns a map with credentials masked for HTTP responses.
|
|
func maskInstanceHTTP(inst store.ChannelInstanceData) map[string]any {
|
|
result := map[string]any{
|
|
"id": inst.ID,
|
|
"name": inst.Name,
|
|
"display_name": inst.DisplayName,
|
|
"channel_type": inst.ChannelType,
|
|
"agent_id": inst.AgentID,
|
|
"config": inst.Config,
|
|
"enabled": inst.Enabled,
|
|
"is_default": store.IsDefaultChannelInstance(inst.Name),
|
|
"has_credentials": len(inst.Credentials) > 0,
|
|
"created_by": inst.CreatedBy,
|
|
"created_at": inst.CreatedAt,
|
|
"updated_at": inst.UpdatedAt,
|
|
}
|
|
|
|
if len(inst.Credentials) > 0 {
|
|
var raw map[string]any
|
|
if json.Unmarshal(inst.Credentials, &raw) == nil {
|
|
masked := make(map[string]any, len(raw))
|
|
for k := range raw {
|
|
masked[k] = "***"
|
|
}
|
|
result["credentials"] = masked
|
|
} else {
|
|
result["credentials"] = map[string]string{}
|
|
}
|
|
} else {
|
|
result["credentials"] = map[string]string{}
|
|
}
|
|
|
|
return result
|
|
}
|
|
|
|
// --- Group file writers ---
|
|
|
|
// resolveAgentID looks up the channel instance and returns its agent_id.
|
|
func (h *ChannelInstancesHandler) resolveAgentID(w http.ResponseWriter, r *http.Request) (uuid.UUID, bool) {
|
|
inst, ok := h.resolveInstance(w, r)
|
|
if !ok {
|
|
return uuid.Nil, false
|
|
}
|
|
return inst.AgentID, true
|
|
}
|
|
|
|
func (h *ChannelInstancesHandler) resolveInstance(w http.ResponseWriter, r *http.Request) (*store.ChannelInstanceData, bool) {
|
|
locale := store.LocaleFromContext(r.Context())
|
|
instID, err := uuid.Parse(r.PathValue("id"))
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidID, "instance"))
|
|
return nil, false
|
|
}
|
|
inst, err := h.store.Get(r.Context(), instID)
|
|
if err != nil {
|
|
writeError(w, http.StatusNotFound, protocol.ErrNotFound, i18n.T(locale, i18n.MsgInstanceNotFound))
|
|
return nil, false
|
|
}
|
|
return inst, true
|
|
}
|
|
|
|
func (h *ChannelInstancesHandler) handleWriterGroups(w http.ResponseWriter, r *http.Request) {
|
|
agentID, ok := h.resolveAgentID(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
perms, err := h.configPermStore.List(r.Context(), agentID, store.ConfigTypeFileWriter, "")
|
|
if err != nil {
|
|
slog.Error("channel_instances.writer_groups", "error", err)
|
|
locale := store.LocaleFromContext(r.Context())
|
|
writeError(w, http.StatusInternalServerError, protocol.ErrInternal, i18n.T(locale, i18n.MsgFailedToList, "writer groups"))
|
|
return
|
|
}
|
|
// Group by scope
|
|
counts := make(map[string]int)
|
|
for _, p := range perms {
|
|
if p.Permission == "allow" {
|
|
counts[p.Scope]++
|
|
}
|
|
}
|
|
type groupInfo struct {
|
|
GroupID string `json:"group_id"`
|
|
WriterCount int `json:"writer_count"`
|
|
}
|
|
groups := make([]groupInfo, 0, len(counts))
|
|
for scope, count := range counts {
|
|
groups = append(groups, groupInfo{GroupID: scope, WriterCount: count})
|
|
}
|
|
writeJSON(w, http.StatusOK, map[string]any{"groups": groups})
|
|
}
|
|
|
|
func (h *ChannelInstancesHandler) handleListWriters(w http.ResponseWriter, r *http.Request) {
|
|
agentID, ok := h.resolveAgentID(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
locale := store.LocaleFromContext(r.Context())
|
|
groupID := r.URL.Query().Get("group_id")
|
|
if groupID == "" {
|
|
writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgRequired, "group_id"))
|
|
return
|
|
}
|
|
perms, err := h.configPermStore.List(r.Context(), agentID, store.ConfigTypeFileWriter, groupID)
|
|
if err != nil {
|
|
slog.Error("channel_instances.list_writers", "error", err)
|
|
writeError(w, http.StatusInternalServerError, protocol.ErrInternal, i18n.T(locale, i18n.MsgFailedToList, "writers"))
|
|
return
|
|
}
|
|
type writerData struct {
|
|
UserID string `json:"user_id"`
|
|
DisplayName *string `json:"display_name,omitempty"`
|
|
Username *string `json:"username,omitempty"`
|
|
}
|
|
writers := make([]writerData, 0, len(perms))
|
|
for _, p := range perms {
|
|
if p.Permission != "allow" {
|
|
continue
|
|
}
|
|
wd := writerData{UserID: p.UserID}
|
|
var meta struct {
|
|
DisplayName string `json:"displayName"`
|
|
Username string `json:"username"`
|
|
}
|
|
if json.Unmarshal(p.Metadata, &meta) == nil {
|
|
if meta.DisplayName != "" {
|
|
wd.DisplayName = &meta.DisplayName
|
|
}
|
|
if meta.Username != "" {
|
|
wd.Username = &meta.Username
|
|
}
|
|
}
|
|
writers = append(writers, wd)
|
|
}
|
|
writeJSON(w, http.StatusOK, map[string]any{"writers": writers})
|
|
}
|
|
|
|
func (h *ChannelInstancesHandler) handleTestWriter(w http.ResponseWriter, r *http.Request) {
|
|
inst, ok := h.resolveInstance(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
locale := store.LocaleFromContext(r.Context())
|
|
var body struct {
|
|
GroupID string `json:"group_id"`
|
|
UserID string `json:"user_id"`
|
|
}
|
|
if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<16)).Decode(&body); err != nil {
|
|
writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidJSON))
|
|
return
|
|
}
|
|
if body.GroupID == "" || body.UserID == "" {
|
|
writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgRequired, "group_id and user_id"))
|
|
return
|
|
}
|
|
channelName, _, groupOK := channels.ParseGroupScope(body.GroupID)
|
|
if !groupOK || channelName != inst.Name {
|
|
writeJSON(w, http.StatusOK, map[string]any{
|
|
"allowed": false,
|
|
"reason": "invalid_group",
|
|
"instance_id": inst.ID.String(),
|
|
"agent_id": inst.AgentID.String(),
|
|
"group_id": body.GroupID,
|
|
"user_id": body.UserID,
|
|
"writer_count": 0,
|
|
})
|
|
return
|
|
}
|
|
|
|
writers, err := h.configPermStore.ListFileWriters(r.Context(), inst.AgentID, body.GroupID)
|
|
if err != nil {
|
|
slog.Error("channel_instances.test_writer", "error", err)
|
|
writeError(w, http.StatusInternalServerError, protocol.ErrInternal, i18n.T(locale, i18n.MsgFailedToList, "writers"))
|
|
return
|
|
}
|
|
allowed := false
|
|
for _, p := range writers {
|
|
if p.Permission == "allow" && p.UserID == body.UserID {
|
|
allowed = true
|
|
break
|
|
}
|
|
}
|
|
reason := "not_writer"
|
|
if allowed {
|
|
reason = "writer"
|
|
} else if len(writers) == 0 {
|
|
reason = "no_writers_configured"
|
|
}
|
|
writeJSON(w, http.StatusOK, map[string]any{
|
|
"allowed": allowed,
|
|
"reason": reason,
|
|
"instance_id": inst.ID.String(),
|
|
"agent_id": inst.AgentID.String(),
|
|
"group_id": body.GroupID,
|
|
"user_id": body.UserID,
|
|
"writer_count": len(writers),
|
|
})
|
|
}
|
|
|
|
func (h *ChannelInstancesHandler) handleAddWriter(w http.ResponseWriter, r *http.Request) {
|
|
agentID, ok := h.resolveAgentID(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
locale := store.LocaleFromContext(r.Context())
|
|
var body struct {
|
|
GroupID string `json:"group_id"`
|
|
UserID string `json:"user_id"`
|
|
DisplayName string `json:"display_name"`
|
|
Username string `json:"username"`
|
|
}
|
|
if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<16)).Decode(&body); err != nil {
|
|
writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidJSON))
|
|
return
|
|
}
|
|
if body.GroupID == "" || body.UserID == "" {
|
|
writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgRequired, "group_id and user_id"))
|
|
return
|
|
}
|
|
meta, _ := json.Marshal(map[string]string{"displayName": body.DisplayName, "username": body.Username})
|
|
// Auto-enrich when caller supplied neither display name nor username.
|
|
// Best-effort — any resolver error leaves the metadata as-is so the
|
|
// store's fallback ({}) applies and the grant still succeeds.
|
|
if body.DisplayName == "" && body.Username == "" {
|
|
if enriched, ok := channels.EnrichFileWriterMetadata(r.Context(), h.memberResolver, body.GroupID, body.UserID); ok {
|
|
meta = enriched
|
|
}
|
|
}
|
|
if err := h.configPermStore.Grant(r.Context(), &store.ConfigPermission{
|
|
AgentID: agentID,
|
|
Scope: body.GroupID,
|
|
ConfigType: store.ConfigTypeFileWriter,
|
|
UserID: body.UserID,
|
|
Permission: "allow",
|
|
Metadata: meta,
|
|
}); err != nil {
|
|
slog.Error("channel_instances.add_writer", "error", err)
|
|
writeError(w, http.StatusInternalServerError, protocol.ErrInternal, i18n.T(locale, i18n.MsgFailedToCreate, "writer", "internal error"))
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusCreated, map[string]string{"status": "added"})
|
|
}
|
|
|
|
func (h *ChannelInstancesHandler) handleRemoveWriter(w http.ResponseWriter, r *http.Request) {
|
|
agentID, ok := h.resolveAgentID(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
locale := store.LocaleFromContext(r.Context())
|
|
userID := r.PathValue("userId")
|
|
groupID := r.URL.Query().Get("group_id")
|
|
if groupID == "" || userID == "" {
|
|
writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgRequired, "group_id and userId"))
|
|
return
|
|
}
|
|
// Prevent removing the last writer (same guard as Telegram /removewriter)
|
|
writers, _ := h.configPermStore.List(r.Context(), agentID, store.ConfigTypeFileWriter, groupID)
|
|
allowCount := 0
|
|
for _, p := range writers {
|
|
if p.Permission == "allow" {
|
|
allowCount++
|
|
}
|
|
}
|
|
if allowCount <= 1 {
|
|
writeError(w, http.StatusConflict, protocol.ErrFailedPrecondition, i18n.T(locale, i18n.MsgCannotRemoveLastWriter))
|
|
return
|
|
}
|
|
if err := h.configPermStore.Revoke(r.Context(), agentID, groupID, store.ConfigTypeFileWriter, userID); err != nil {
|
|
slog.Error("channel_instances.remove_writer", "error", err)
|
|
writeError(w, http.StatusInternalServerError, protocol.ErrInternal, i18n.T(locale, i18n.MsgFailedToDelete, "writer", "internal error"))
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, map[string]string{"status": "removed"})
|
|
}
|
|
|
|
// --- Channel contacts ---
|
|
|
|
func (h *ChannelInstancesHandler) handleListContacts(w http.ResponseWriter, r *http.Request) {
|
|
opts := store.ContactListOpts{
|
|
Limit: 50,
|
|
Offset: 0,
|
|
}
|
|
|
|
if v := r.URL.Query().Get("search"); v != "" {
|
|
opts.Search = v
|
|
}
|
|
if v := r.URL.Query().Get("channel_type"); v != "" {
|
|
opts.ChannelType = v
|
|
}
|
|
if v := r.URL.Query().Get("channel_instance"); v != "" {
|
|
opts.ChannelInstance = v
|
|
}
|
|
if v := r.URL.Query().Get("peer_kind"); v != "" {
|
|
opts.PeerKind = v
|
|
}
|
|
if v := r.URL.Query().Get("contact_type"); v != "" {
|
|
opts.ContactType = v
|
|
}
|
|
if v := r.URL.Query().Get("limit"); v != "" {
|
|
if n, err := strconv.Atoi(v); err == nil && n > 0 && n <= 200 {
|
|
opts.Limit = n
|
|
}
|
|
}
|
|
if v := r.URL.Query().Get("offset"); v != "" {
|
|
if n, err := strconv.Atoi(v); err == nil && n >= 0 {
|
|
opts.Offset = n
|
|
}
|
|
}
|
|
|
|
contacts, err := h.contactStore.ListContacts(r.Context(), opts)
|
|
if err != nil {
|
|
slog.Error("contacts.list", "error", err)
|
|
locale := store.LocaleFromContext(r.Context())
|
|
writeError(w, http.StatusInternalServerError, protocol.ErrInternal, i18n.T(locale, i18n.MsgFailedToList, "contacts"))
|
|
return
|
|
}
|
|
if contacts == nil {
|
|
contacts = []store.ChannelContact{}
|
|
}
|
|
|
|
total, countErr := h.contactStore.CountContacts(r.Context(), opts)
|
|
if countErr != nil {
|
|
slog.Warn("contacts.count", "error", countErr)
|
|
}
|
|
|
|
writeJSON(w, http.StatusOK, map[string]any{
|
|
"contacts": contacts,
|
|
"total": total,
|
|
"limit": opts.Limit,
|
|
"offset": opts.Offset,
|
|
})
|
|
}
|
|
|
|
func (h *ChannelInstancesHandler) handleResolveContacts(w http.ResponseWriter, r *http.Request) {
|
|
idsParam := r.URL.Query().Get("ids")
|
|
if idsParam == "" {
|
|
writeJSON(w, http.StatusOK, map[string]any{"contacts": map[string]any{}})
|
|
return
|
|
}
|
|
|
|
ids := strings.Split(idsParam, ",")
|
|
if len(ids) > 100 {
|
|
ids = ids[:100]
|
|
}
|
|
|
|
result, err := h.contactStore.GetContactsBySenderIDs(r.Context(), ids)
|
|
if err != nil {
|
|
slog.Error("contacts.resolve", "error", err)
|
|
locale := store.LocaleFromContext(r.Context())
|
|
writeError(w, http.StatusInternalServerError, protocol.ErrInternal, i18n.T(locale, i18n.MsgFailedToList, "contacts"))
|
|
return
|
|
}
|
|
|
|
writeJSON(w, http.StatusOK, map[string]any{"contacts": result})
|
|
}
|
|
|
|
// isValidChannelType checks if the channel type is supported.
|
|
//
|
|
// Keep this list in sync with the WS twin in
|
|
// internal/gateway/methods/channel_instances.go and with CHANNEL_TYPES in
|
|
// ui/web/src/constants/channels.ts.
|
|
func isValidChannelType(ct string) bool {
|
|
switch ct {
|
|
case "telegram", "discord", "slack", "whatsapp", "zalo_oa", "zalo_personal", "feishu", "facebook", "pancake", "bitrix24":
|
|
return true
|
|
}
|
|
return false
|
|
}
|