package channel import ( "context" "errors" "fmt" "log/slog" "strings" "time" ) type connectionEntry struct { config ChannelConfig connection Connection } func (m *Manager) refresh(ctx context.Context) { // Serialize refresh calls so concurrent callers wait instead of silently skipping. m.refreshMu.Lock() defer m.refreshMu.Unlock() if m.service == nil { return } configs := make([]ChannelConfig, 0) for _, channelType := range m.registry.Types() { items, err := m.service.ListConfigsByType(ctx, channelType) if err != nil { if m.logger != nil { m.logger.Error("list configs failed", slog.String("channel", channelType.String()), slog.Any("error", err)) } continue } configs = append(configs, items...) } m.reconcile(ctx, configs) } func (m *Manager) reconcile(ctx context.Context, configs []ChannelConfig) { active := map[string]ChannelConfig{} for _, cfg := range configs { if cfg.ID == "" || cfg.Disabled { continue } active[cfg.ID] = cfg if err := m.ensureConnection(ctx, cfg); err != nil { m.markConnectionStatus(cfg, false, err) if m.logger != nil { m.logger.Error( "adapter start failed", slog.String("bot_id", cfg.BotID), slog.String("channel", cfg.ChannelType.String()), slog.String("config_id", cfg.ID), slog.Any("error", err), ) } } } m.mu.Lock() defer m.mu.Unlock() for id, entry := range m.connections { if _, ok := active[id]; ok { continue } if entry != nil && entry.connection != nil { if m.logger != nil { m.logger.Info( "adapter stop", slog.String("bot_id", entry.config.BotID), slog.String("channel", entry.config.ChannelType.String()), slog.String("config_id", id), ) } if err := entry.connection.Stop(ctx); err != nil && !errors.Is(err, ErrStopNotSupported) && m.logger != nil { m.logger.Warn( "adapter stop failed", slog.String("bot_id", entry.config.BotID), slog.String("channel", entry.config.ChannelType.String()), slog.String("config_id", id), slog.Any("error", err), ) } } delete(m.connections, id) delete(m.connectionMeta, id) } for id := range m.connectionMeta { if _, ok := active[id]; !ok { delete(m.connectionMeta, id) } } } func (m *Manager) ensureConnection(ctx context.Context, cfg ChannelConfig) error { _, ok := m.registry.GetReceiver(cfg.ChannelType) if !ok { m.markConnectionStatus(cfg, false, fmt.Errorf("receiver not available")) return nil } m.mu.Lock() entry := m.connections[cfg.ID] // Config unchanged — nothing to do. if entry != nil && !entry.config.UpdatedAt.Before(cfg.UpdatedAt) { running := entry.connection != nil && entry.connection.Running() m.setConnectionStatusLocked(entry.config, running, nil) m.mu.Unlock() return nil } // Need to stop existing connection before starting a new one. // Keep the lock to prevent another goroutine from starting a duplicate. var oldConn Connection if entry != nil { oldConn = entry.connection delete(m.connections, cfg.ID) } m.mu.Unlock() if oldConn != nil { if m.logger != nil { m.logger.Info( "adapter restart", slog.String("bot_id", cfg.BotID), slog.String("channel", cfg.ChannelType.String()), slog.String("config_id", cfg.ID), ) } if err := oldConn.Stop(ctx); err != nil { if errors.Is(err, ErrStopNotSupported) { if m.logger != nil { m.logger.Warn( "adapter restart skipped", slog.String("bot_id", cfg.BotID), slog.String("channel", cfg.ChannelType.String()), slog.String("config_id", cfg.ID), ) } // Re-insert the entry since we can't restart it. m.mu.Lock() if _, exists := m.connections[cfg.ID]; !exists { m.connections[cfg.ID] = entry running := entry != nil && entry.connection != nil && entry.connection.Running() m.setConnectionStatusLocked(entry.config, running, nil) } m.mu.Unlock() return nil } m.markConnectionStatus(cfg, false, err) return err } } receiver, ok := m.registry.GetReceiver(cfg.ChannelType) if !ok { m.markConnectionStatus(cfg, false, fmt.Errorf("receiver not available")) return nil } // Double-check: another goroutine may have already started a connection // for this config while we were stopping the old one. m.mu.Lock() if existing, ok := m.connections[cfg.ID]; ok && existing != nil { running := existing.connection != nil && existing.connection.Running() m.setConnectionStatusLocked(existing.config, running, nil) m.mu.Unlock() return nil } m.mu.Unlock() if m.logger != nil { m.logger.Info( "adapter start", slog.String("bot_id", cfg.BotID), slog.String("channel", cfg.ChannelType.String()), slog.String("config_id", cfg.ID), ) } handler := m.handleInbound for i := len(m.middlewares) - 1; i >= 0; i-- { handler = m.middlewares[i](handler) } connectCtx := context.Background() if ctx != nil { // Decouple long-lived adapter connections from short-lived request contexts. connectCtx = context.WithoutCancel(ctx) } conn, err := receiver.Connect(connectCtx, cfg, handler) if err != nil { m.markConnectionStatus(cfg, false, err) return err } m.mu.Lock() // Final check: if another goroutine raced and inserted first, stop our new // connection and keep the existing one. if existing, ok := m.connections[cfg.ID]; ok && existing != nil { running := existing.connection != nil && existing.connection.Running() m.setConnectionStatusLocked(existing.config, running, nil) m.mu.Unlock() _ = conn.Stop(context.Background()) return nil } m.connections[cfg.ID] = &connectionEntry{ config: cfg, connection: conn, } m.setConnectionStatusLocked(cfg, true, nil) m.mu.Unlock() return nil } // EnsureConnection starts, restarts, or stops the connection for the given config. // Disabled configs are stopped and removed; enabled configs are started or restarted. func (m *Manager) EnsureConnection(ctx context.Context, cfg ChannelConfig) error { if cfg.ID == "" { return fmt.Errorf("config id is required") } if cfg.Disabled { return m.removeConnection(ctx, cfg.ID) } return m.ensureConnection(ctx, cfg) } // RemoveConnection stops and removes connections matching the given bot and channel type. func (m *Manager) RemoveConnection(ctx context.Context, botID string, channelType ChannelType) { botID = strings.TrimSpace(botID) if botID == "" { return } m.mu.Lock() defer m.mu.Unlock() for id, entry := range m.connections { if entry == nil || entry.config.BotID != botID || entry.config.ChannelType != channelType { continue } if entry.connection != nil { if m.logger != nil { m.logger.Info( "connection remove", slog.String("bot_id", botID), slog.String("channel", channelType.String()), slog.String("config_id", id), ) } if err := entry.connection.Stop(ctx); err != nil && !errors.Is(err, ErrStopNotSupported) && m.logger != nil { m.logger.Warn( "connection stop failed", slog.String("bot_id", botID), slog.String("channel", channelType.String()), slog.String("config_id", id), slog.Any("error", err), ) } } delete(m.connections, id) delete(m.connectionMeta, id) } } func (m *Manager) removeConnection(ctx context.Context, configID string) error { m.mu.Lock() entry := m.connections[configID] if entry == nil { delete(m.connectionMeta, configID) m.mu.Unlock() return nil } delete(m.connections, configID) delete(m.connectionMeta, configID) m.mu.Unlock() if entry.connection != nil { if m.logger != nil { m.logger.Info( "connection remove", slog.String("bot_id", entry.config.BotID), slog.String("channel", entry.config.ChannelType.String()), slog.String("config_id", configID), ) } if err := entry.connection.Stop(ctx); err != nil && !errors.Is(err, ErrStopNotSupported) { if m.logger != nil { m.logger.Warn( "connection stop failed", slog.String("bot_id", entry.config.BotID), slog.String("channel", entry.config.ChannelType.String()), slog.String("config_id", configID), slog.Any("error", err), ) } return err } } return nil } func (m *Manager) stopAll(ctx context.Context) { m.mu.Lock() defer m.mu.Unlock() for id, entry := range m.connections { if entry != nil && entry.connection != nil { if m.logger != nil { m.logger.Info( "adapter stop", slog.String("bot_id", entry.config.BotID), slog.String("channel", entry.config.ChannelType.String()), slog.String("config_id", id), ) } if err := entry.connection.Stop(ctx); err != nil && !errors.Is(err, ErrStopNotSupported) && m.logger != nil { m.logger.Warn( "adapter stop failed", slog.String("bot_id", entry.config.BotID), slog.String("channel", entry.config.ChannelType.String()), slog.String("config_id", id), slog.Any("error", err), ) } } delete(m.connections, id) delete(m.connectionMeta, id) } } // Stop terminates the connection identified by the given config ID. func (m *Manager) Stop(ctx context.Context, configID string) error { configID = strings.TrimSpace(configID) if configID == "" { return fmt.Errorf("config id is required") } m.mu.Lock() entry := m.connections[configID] m.mu.Unlock() if entry == nil || entry.connection == nil { return nil } err := entry.connection.Stop(ctx) if err != nil { return err } m.markConnectionStatus(entry.config, false, nil) return nil } // StopByBot terminates all connections belonging to the given bot. func (m *Manager) StopByBot(ctx context.Context, botID string) error { botID = strings.TrimSpace(botID) if botID == "" { return fmt.Errorf("bot id is required") } m.mu.Lock() defer m.mu.Unlock() for id, entry := range m.connections { if entry != nil && entry.config.BotID == botID { if entry.connection != nil { _ = entry.connection.Stop(ctx) } delete(m.connections, id) delete(m.connectionMeta, id) } } return nil } func (m *Manager) markConnectionStatus(cfg ChannelConfig, running bool, checkErr error) { m.mu.Lock() defer m.mu.Unlock() m.setConnectionStatusLocked(cfg, running, checkErr) } func (m *Manager) setConnectionStatusLocked(cfg ChannelConfig, running bool, checkErr error) { if strings.TrimSpace(cfg.ID) == "" { return } if m.connectionMeta == nil { m.connectionMeta = map[string]ConnectionStatus{} } previous, hasPrevious := m.connectionMeta[cfg.ID] status := ConnectionStatus{ ConfigID: cfg.ID, BotID: cfg.BotID, ChannelType: cfg.ChannelType, Running: running, UpdatedAt: time.Now().UTC(), } if checkErr != nil { status.LastError = checkErr.Error() } m.connectionMeta[cfg.ID] = status if m.logger != nil { if checkErr != nil && (!hasPrevious || previous.LastError != status.LastError || previous.Running != status.Running) { m.logger.Warn( "connection health check failed", slog.String("bot_id", cfg.BotID), slog.String("channel", cfg.ChannelType.String()), slog.String("config_id", cfg.ID), slog.Any("error", checkErr), ) } if running && hasPrevious && strings.TrimSpace(previous.LastError) != "" { m.logger.Info( "connection health recovered", slog.String("bot_id", cfg.BotID), slog.String("channel", cfg.ChannelType.String()), slog.String("config_id", cfg.ID), ) } } }