diff --git a/bridge.go b/bridge.go new file mode 100644 index 0000000..ec53eb6 --- /dev/null +++ b/bridge.go @@ -0,0 +1,170 @@ +package main + +import ( + "context" + "fmt" + "log" + "sync" + + "maunium.net/go/mautrix" + "maunium.net/go/mautrix/appservice" + "maunium.net/go/mautrix/event" + "maunium.net/go/mautrix/id" +) + +type Bridge struct { + as *appservice.AppService + db *Database + ntfy *NtfyClient + cfg *Config + + mu sync.Mutex + listeners map[string]context.CancelFunc +} + +func NewBridge(as *appservice.AppService, db *Database, ntfy *NtfyClient, cfg *Config) *Bridge { + return &Bridge{ + as: as, + db: db, + ntfy: ntfy, + cfg: cfg, + listeners: make(map[string]context.CancelFunc), + } +} + +func (b *Bridge) StartExistingBridges(ctx context.Context) { + bridges, err := b.db.GetAllBridges() + if err != nil { + log.Fatalf("Failed to load bridges: %v", err) + } + + botIntent := b.as.BotIntent() + for topic, roomID := range bridges { + log.Printf("[startup] Restoring listener: topic=%q room=%s", topic, roomID) + b.startListener(topic, roomID, botIntent) + } + + log.Printf("[startup] Restored %d bridge(s)", len(bridges)) +} + +func (b *Bridge) CreateBridge(ctx context.Context, sender id.UserID, topicName string) (id.RoomID, error) { + botIntent := b.as.BotIntent() + + if existing, err := b.db.GetBridgeByTopic(topicName); err != nil { + return "", fmt.Errorf("database error: %w", err) + } else if existing != "" { + return "", fmt.Errorf("topic '%s' is already bridged to %s", topicName, existing) + } + + spaceRoomID, err := b.ensureSpace(ctx, sender) + if err != nil { + return "", fmt.Errorf("space setup: %w", err) + } + + roomResp, err := botIntent.Client.CreateRoom(ctx, &mautrix.ReqCreateRoom{ + Name: topicName, + Topic: fmt.Sprintf("ntfy topic: %s", topicName), + Preset: "private_chat", + Invite: []id.UserID{sender}, + }) + if err != nil { + return "", fmt.Errorf("create room: %w", err) + } + bridgeRoomID := roomResp.RoomID + + if _, err := botIntent.SendStateEvent(ctx, id.RoomID(spaceRoomID), event.StateSpaceChild, string(bridgeRoomID), &event.SpaceChildEventContent{ + Via: []string{b.cfg.Homeserver.Domain}, + }); err != nil { + log.Printf("[bridge] failed to add room to space: %v", err) + } + + userIntent := b.as.Intent(sender) + userIntent.IsCustomPuppet = true + if err := userIntent.EnsureJoined(ctx, bridgeRoomID); err != nil { + log.Printf("[bridge] double puppet join failed: %v", err) + } + + if err := b.db.SetBridge(topicName, string(bridgeRoomID)); err != nil { + return "", fmt.Errorf("save bridge: %w", err) + } + + b.startListener(topicName, string(bridgeRoomID), botIntent) + + return bridgeRoomID, nil +} + +func (b *Bridge) DeleteBridge(topicName string) error { + b.mu.Lock() + if cancel, ok := b.listeners[topicName]; ok { + cancel() + delete(b.listeners, topicName) + } + b.mu.Unlock() + + return b.db.DeleteBridge(topicName) +} + +func (b *Bridge) PublishToNtfy(topic, message string) error { + return b.ntfy.PublishToNtfy(topic, message) +} + +func (b *Bridge) GetBridgeByRoom(roomID string) (string, error) { + return b.db.GetBridgeByRoom(roomID) +} + +func (b *Bridge) GetAllBridges() (map[string]string, error) { + return b.db.GetAllBridges() +} + +func (b *Bridge) HasPermission(userID string) bool { + return b.cfg.Bridge.HasPermission(userID, "admin") +} + +func (b *Bridge) startListener(topic, roomID string, intent *appservice.IntentAPI) { + b.mu.Lock() + defer b.mu.Unlock() + + if _, exists := b.listeners[topic]; exists { + return + } + + ctx, cancel := context.WithCancel(context.Background()) + b.listeners[topic] = cancel + + go b.ntfy.Listen(ctx, topic, roomID, intent) +} + +func (b *Bridge) ensureSpace(ctx context.Context, sender id.UserID) (string, error) { + spaceRoomID, err := b.db.GetSpaceRoomID(string(sender)) + if err != nil { + return "", err + } + if spaceRoomID != "" { + return spaceRoomID, nil + } + + resp, err := b.as.BotIntent().Client.CreateRoom(ctx, &mautrix.ReqCreateRoom{ + Name: "Notifications", + Preset: "private_chat", + CreationContent: map[string]interface{}{ + "type": "m.space", + }, + Invite: []id.UserID{sender}, + }) + if err != nil { + return "", err + } + spaceRoomID = string(resp.RoomID) + + if err := b.db.SetSpaceRoomID(string(sender), spaceRoomID); err != nil { + return "", err + } + + userIntent := b.as.Intent(sender) + userIntent.IsCustomPuppet = true + if err := userIntent.EnsureJoined(ctx, id.RoomID(spaceRoomID)); err != nil { + log.Printf("[bridge] double puppet join space failed: %v", err) + } + + return spaceRoomID, nil +} diff --git a/config.go b/config.go index 2d9ef21..ebffcfa 100644 --- a/config.go +++ b/config.go @@ -1,6 +1,7 @@ package main import ( + "fmt" "os" "gopkg.in/yaml.v3" @@ -13,11 +14,10 @@ type HomeserverConfig struct { type AppserviceConfig struct { ID string `yaml:"id"` + Address string `yaml:"address"` + BotLocalpart string `yaml:"bot_localpart"` ASToken string `yaml:"as_token"` HSToken string `yaml:"hs_token"` - BotLocalpart string `yaml:"bot_localpart"` - Address string `yaml:"address"` - Port uint16 `yaml:"port"` } type BridgeConfig struct { @@ -40,32 +40,37 @@ func LoadConfig(path string) (*Config, error) { if err = yaml.Unmarshal(data, cfg); err != nil { return nil, err } - if cfg.Bridge.Permissions == nil { - cfg.Bridge.Permissions = make(map[string]string) - } - if cfg.Bridge.NtfyURL == "" { - cfg.Bridge.NtfyURL = "https://ntfy.sh" - } - if cfg.Appservice.Address == "" { - cfg.Appservice.Address = "0.0.0.0" - } - if cfg.Appservice.Port == 0 { - cfg.Appservice.Port = 8080 - } + cfg.applyDefaults() return cfg, nil } -func (b *BridgeConfig) PermissionLevel(id string) string { - if level, ok := b.Permissions[id]; ok { - return level +func (c *Config) applyDefaults() { + if c.Bridge.Permissions == nil { + c.Bridge.Permissions = make(map[string]string) + } + if c.Bridge.NtfyURL == "" { + c.Bridge.NtfyURL = "https://ntfy.sh" + } + if c.Appservice.ID == "" { + c.Appservice.ID = "ntfy-bridge" + } + if c.Appservice.Address == "" { + c.Appservice.Address = "0.0.0.0" } - return "" } -func (b *BridgeConfig) HasPermission(userID, requiredLevel string) bool { - level := b.PermissionLevel(userID) +func (c *Config) Save(path string) error { + data, err := yaml.Marshal(c) + if err != nil { + return fmt.Errorf("marshal config: %w", err) + } + return os.WriteFile(path, data, 0600) +} + +func (c *BridgeConfig) HasPermission(userID, requiredLevel string) bool { + level := c.PermissionLevel(userID) if level == "" { - level = b.PermissionLevel("*") + level = c.PermissionLevel("*") } if level == "" { return false @@ -75,3 +80,10 @@ func (b *BridgeConfig) HasPermission(userID, requiredLevel string) bool { } return true } + +func (c *BridgeConfig) PermissionLevel(id string) string { + if level, ok := c.Permissions[id]; ok { + return level + } + return "" +} diff --git a/database.go b/database.go index 241911a..71d280d 100644 --- a/database.go +++ b/database.go @@ -104,3 +104,8 @@ func (d *Database) SetBridge(ntfyTopic, matrixRoomID string) error { ) return err } + +func (d *Database) DeleteBridge(ntfyTopic string) error { + _, err := d.db.Exec("DELETE FROM bridges WHERE ntfy_topic = ?", ntfyTopic) + return err +} diff --git a/handler.go b/handler.go new file mode 100644 index 0000000..35e4d08 --- /dev/null +++ b/handler.go @@ -0,0 +1,185 @@ +package main + +import ( + "context" + "fmt" + "log" + "strings" + + "maunium.net/go/mautrix/event" + "maunium.net/go/mautrix/id" +) + +type MatrixHandler struct { + bridge *Bridge + ghostPrefix string +} + +func NewMatrixHandler(bridge *Bridge, ghostPrefix string) *MatrixHandler { + return &MatrixHandler{ + bridge: bridge, + ghostPrefix: ghostPrefix, + } +} + +func (h *MatrixHandler) HandleEvent(ctx context.Context, evt *event.Event) { + if evt.Sender == h.bridge.as.BotMXID() { + return + } + if h.ghostPrefix != "" && strings.HasPrefix(string(evt.Sender), h.ghostPrefix) { + return + } + + switch evt.Type { + case event.EventMessage: + h.onMessage(ctx, evt) + case event.StateMember: + h.onMembership(ctx, evt) + } +} + +func (h *MatrixHandler) onMessage(ctx context.Context, evt *event.Event) { + msg := evt.Content.AsMessage() + if msg == nil { + return + } + + text := strings.TrimSpace(msg.Body) + if text == "" { + return + } + + cmd, arg := parseCommand(text) + + switch cmd { + case "help", "": + h.cmdHelp(ctx, evt) + + case "list": + h.cmdList(ctx, evt) + + case "create": + h.cmdCreate(ctx, evt, arg) + + case "delete": + h.cmdDelete(ctx, evt, arg) + + default: + if topic, _ := h.bridge.GetBridgeByRoom(string(evt.RoomID)); topic != "" { + h.forwardToNtfy(ctx, evt, topic) + } else { + h.reply(ctx, evt.RoomID, "Type 'help' to see available commands.") + } + } +} + +func (h *MatrixHandler) onMembership(ctx context.Context, evt *event.Event) { + member := evt.Content.AsMember() + if member == nil { + return + } + + stateKey := evt.GetStateKey() + if stateKey == "" { + return + } + + if member.Membership == event.MembershipInvite && id.UserID(stateKey) == h.bridge.as.BotMXID() { + if err := h.bridge.as.BotIntent().EnsureJoined(ctx, evt.RoomID); err != nil { + log.Printf("[matrix] auto-join room %s failed: %v", evt.RoomID, err) + } + } +} + +func (h *MatrixHandler) cmdHelp(ctx context.Context, evt *event.Event) { + h.reply(ctx, evt.RoomID, + "ntfy-Matrix Bridge\n\n"+ + " bridge create — Bridge an ntfy topic to a new room\n"+ + " bridge list — List all active bridges\n"+ + " bridge delete — Remove a bridge\n"+ + " help — Show this message\n\n"+ + "Bridged rooms forward messages between Matrix and ntfy.", + ) +} + +func (h *MatrixHandler) cmdList(ctx context.Context, evt *event.Event) { + bridges, err := h.bridge.GetAllBridges() + if err != nil { + log.Printf("[matrix] list bridges: %v", err) + h.reply(ctx, evt.RoomID, "Database error.") + return + } + if len(bridges) == 0 { + h.reply(ctx, evt.RoomID, "No active bridges.") + return + } + + var b strings.Builder + b.WriteString("Active bridges:\n") + for topic, roomID := range bridges { + fmt.Fprintf(&b, " %s → %s\n", topic, roomID) + } + h.reply(ctx, evt.RoomID, strings.TrimRight(b.String(), "\n")) +} + +func (h *MatrixHandler) cmdCreate(ctx context.Context, evt *event.Event, topicName string) { + if topicName == "" { + h.reply(ctx, evt.RoomID, "Usage: bridge create ") + return + } + if !h.bridge.HasPermission(string(evt.Sender)) { + h.reply(ctx, evt.RoomID, "Admin access required.") + return + } + + roomID, err := h.bridge.CreateBridge(ctx, evt.Sender, topicName) + if err != nil { + h.reply(ctx, evt.RoomID, fmt.Sprintf("Failed: %v", err)) + return + } + + h.reply(ctx, evt.RoomID, fmt.Sprintf("Bridge created!\n Topic: %s\n Room: %s", topicName, roomID)) +} + +func (h *MatrixHandler) cmdDelete(ctx context.Context, evt *event.Event, topicName string) { + if topicName == "" { + h.reply(ctx, evt.RoomID, "Usage: bridge delete ") + return + } + if !h.bridge.HasPermission(string(evt.Sender)) { + h.reply(ctx, evt.RoomID, "Admin access required.") + return + } + + if err := h.bridge.DeleteBridge(topicName); err != nil { + log.Printf("[matrix] delete bridge: %v", err) + h.reply(ctx, evt.RoomID, "Failed to delete bridge.") + return + } + + h.reply(ctx, evt.RoomID, fmt.Sprintf("Bridge for '%s' deleted.", topicName)) +} + +func (h *MatrixHandler) forwardToNtfy(ctx context.Context, evt *event.Event, topic string) { + if err := h.bridge.PublishToNtfy(topic, evt.Content.AsMessage().Body); err != nil { + log.Printf("[matrix] publish to ntfy: %v", err) + h.reply(ctx, evt.RoomID, fmt.Sprintf("Failed to publish: %v", err)) + } +} + +func (h *MatrixHandler) reply(ctx context.Context, roomID id.RoomID, text string) { + h.bridge.as.BotIntent().SendNotice(ctx, roomID, text) +} + +func parseCommand(text string) (cmd, arg string) { + if !strings.HasPrefix(text, "bridge ") { + return "", text + } + rest := strings.TrimSpace(strings.TrimPrefix(text, "bridge ")) + parts := strings.SplitN(rest, " ", 2) + cmd = parts[0] + if len(parts) > 1 { + arg = strings.TrimSpace(parts[1]) + } + return +} diff --git a/main.go b/main.go index 1f3d933..eee8835 100644 --- a/main.go +++ b/main.go @@ -6,24 +6,95 @@ import ( "fmt" "log" "os" + "regexp" "maunium.net/go/mautrix/appservice" "maunium.net/go/mautrix/event" ) +var ( + Version = "dev" + BuildTime = "unknown" +) + func main() { configPath := flag.String("config", "config.yaml", "Path to configuration file") dbPath := flag.String("database", "ntfy-bridge.db", "Path to SQLite database file") + genReg := flag.Bool("g", false, "Generate registration file and exit") + regPath := flag.String("r", "registration.yaml", "Path for the registration file") + showVersion := flag.Bool("v", false, "Print version and exit") flag.Parse() + if *showVersion { + fmt.Printf("mautrix-ntfy %s (built %s)\n", Version, BuildTime) + return + } + cfg, err := LoadConfig(*configPath) if err != nil { - log.Fatalf("Failed to load config: %v", err) + log.Fatalf("Load config: %v", err) } - db, err := NewDatabase(*dbPath) + if *genReg { + if err := generateRegistration(cfg, *configPath, *regPath); err != nil { + log.Fatalf("Generate registration: %v", err) + } + return + } + + if err := run(cfg, *configPath, *dbPath); err != nil { + log.Fatalf("Fatal: %v", err) + } +} + +func generateRegistration(cfg *Config, configPath, regPath string) error { + if cfg.Homeserver.Domain == "" || cfg.Homeserver.Domain == "example.com" { + return fmt.Errorf("set homeserver.domain in %s first", configPath) + } + if cfg.Appservice.BotLocalpart == "" { + return fmt.Errorf("set appservice.bot_localpart in %s first", configPath) + } + if cfg.Appservice.Address == "" { + return fmt.Errorf("set appservice.address in %s first", configPath) + } + + reg := appservice.CreateRegistration() + reg.ID = cfg.Appservice.ID + reg.URL = cfg.Appservice.Address + reg.SenderLocalpart = cfg.Appservice.BotLocalpart + rateLimited := false + reg.RateLimited = &rateLimited + + botRegex := regexp.MustCompile(fmt.Sprintf("^@%s:%s$", + regexp.QuoteMeta(cfg.Appservice.BotLocalpart), + regexp.QuoteMeta(cfg.Homeserver.Domain))) + reg.Namespaces.UserIDs.Register(botRegex, true) + + ghostRegex := regexp.MustCompile(fmt.Sprintf("^@%s_[^:]*:%s$", + regexp.QuoteMeta(cfg.Appservice.BotLocalpart), + regexp.QuoteMeta(cfg.Homeserver.Domain))) + reg.Namespaces.UserIDs.Register(ghostRegex, true) + + if err := reg.Save(regPath); err != nil { + return fmt.Errorf("save registration: %w", err) + } + + cfg.Appservice.ASToken = reg.AppToken + cfg.Appservice.HSToken = reg.ServerToken + if err := cfg.Save(configPath); err != nil { + return fmt.Errorf("save config with tokens: %w", err) + } + + fmt.Printf("Registration saved to %s\n", regPath) + fmt.Printf("Config updated at %s (tokens filled in)\n", configPath) + fmt.Println("Copy the registration file to your homeserver and restart it.") + return nil +} + +func run(cfg *Config, configPath, dbPath string) error { + db, err := NewDatabase(dbPath) if err != nil { - log.Fatalf("Failed to open database: %v", err) + return fmt.Errorf("open database: %w", err) } defer db.Close() @@ -38,47 +109,29 @@ func main() { Registration: reg, HomeserverDomain: cfg.Homeserver.Domain, HomeserverURL: cfg.Homeserver.URL, - HostConfig: appservice.HostConfig{Hostname: cfg.Appservice.Address, Port: cfg.Appservice.Port}, + HostConfig: appservice.HostConfig{Hostname: cfg.Appservice.Address, Port: 8080}, }) if err != nil { - log.Fatalf("Failed to create appservice: %v", err) + return fmt.Errorf("create appservice: %w", err) } - ghostPrefix := fmt.Sprintf("@%s_", cfg.Appservice.BotLocalpart) - - handler := &MatrixHandler{ - AS: as, - DB: db, - Ntfy: NewNtfyClient(cfg.Bridge.NtfyURL), - Config: &cfg.Bridge, - HSDomain: cfg.Homeserver.Domain, - GhostPrefix: ghostPrefix, - } + ntfy := NewNtfyClient(cfg.Bridge.NtfyURL) + bridge := NewBridge(as, db, ntfy, cfg) + handler := NewMatrixHandler(bridge, fmt.Sprintf("@%s_", cfg.Appservice.BotLocalpart)) ep := appservice.NewEventProcessor(as) ep.On(event.EventMessage, handler.HandleEvent) ep.On(event.StateMember, handler.HandleEvent) - bridges, err := db.GetAllBridges() - if err != nil { - log.Fatalf("Failed to load bridges: %v", err) - } - - botIntent := as.BotIntent() - for topic, roomID := range bridges { - log.Printf("[startup] Re-establishing ntfy listener for topic %q -> room %s", topic, roomID) - handler.Ntfy.StartNtfyListener(topic, roomID, botIntent, as.BotMXID()) - } - ctx, cancel := context.WithCancel(context.Background()) defer cancel() ep.Start(ctx) + bridge.StartExistingBridges(ctx) - log.Printf("[main] Starting appservice on %s:%d", cfg.Appservice.Address, cfg.Appservice.Port) - log.Printf("[main] ntfy base URL: %s", cfg.Bridge.NtfyURL) - log.Printf("[main] Loaded %d existing bridge(s)", len(bridges)) + log.Printf("mautrix-ntfy %s starting on %s:8080", Version, cfg.Appservice.Address) as.Start() _ = os.Stdout + return nil } diff --git a/matrix_handler.go b/matrix_handler.go deleted file mode 100644 index f95d2fc..0000000 --- a/matrix_handler.go +++ /dev/null @@ -1,205 +0,0 @@ -package main - -import ( - "context" - "fmt" - "log" - "strings" - - "maunium.net/go/mautrix" - "maunium.net/go/mautrix/appservice" - "maunium.net/go/mautrix/event" - "maunium.net/go/mautrix/id" -) - -type MatrixHandler struct { - AS *appservice.AppService - DB *Database - Ntfy *NtfyClient - Config *BridgeConfig - HSDomain string - GhostPrefix string -} - -func (h *MatrixHandler) HandleEvent(ctx context.Context, evt *event.Event) { - if evt.Sender == h.AS.BotMXID() { - return - } - if h.GhostPrefix != "" && strings.HasPrefix(string(evt.Sender), h.GhostPrefix) { - return - } - - switch evt.Type { - case event.EventMessage: - h.handleMessage(ctx, evt) - case event.StateMember: - h.handleMembership(ctx, evt) - } -} - -func (h *MatrixHandler) handleMessage(ctx context.Context, evt *event.Event) { - msg := evt.Content.AsMessage() - if msg == nil { - return - } - - text := strings.TrimSpace(msg.Body) - - if strings.HasPrefix(text, "bridge ") { - h.handleCommand(ctx, evt, text) - return - } - - topic, err := h.DB.GetBridgeByRoom(string(evt.RoomID)) - if err != nil { - log.Printf("[matrix] db lookup error for room %s: %v", evt.RoomID, err) - return - } - if topic == "" { - return - } - - if err := h.Ntfy.PublishToNtfy(topic, text); err != nil { - botIntent := h.AS.BotIntent() - botIntent.SendNotice(ctx, evt.RoomID, fmt.Sprintf("Failed to publish to ntfy: %v", err)) - log.Printf("[matrix] publish to ntfy failed: %v", err) - } -} - -func (h *MatrixHandler) handleCommand(ctx context.Context, evt *event.Event, text string) { - parts := strings.SplitN(text, " ", 3) - if len(parts) < 2 { - return - } - - switch parts[1] { - case "create": - topicName := "" - if len(parts) > 2 { - topicName = strings.TrimSpace(parts[2]) - } - h.handleCreateBridge(ctx, evt, topicName) - default: - botIntent := h.AS.BotIntent() - botIntent.SendNotice(ctx, evt.RoomID, "Available commands:\n bridge create — Bridge an ntfy topic to a new Matrix room") - } -} - -func (h *MatrixHandler) handleCreateBridge(ctx context.Context, evt *event.Event, topicName string) { - botIntent := h.AS.BotIntent() - - if topicName == "" { - botIntent.SendNotice(ctx, evt.RoomID, "Usage: bridge create ") - return - } - - if !h.Config.HasPermission(string(evt.Sender), "admin") { - botIntent.SendNotice(ctx, evt.RoomID, "You do not have permission to create bridges. Admin access required.") - return - } - - existing, err := h.DB.GetBridgeByTopic(topicName) - if err != nil { - log.Printf("[matrix] db error checking topic: %v", err) - botIntent.SendNotice(ctx, evt.RoomID, "Internal database error.") - return - } - if existing != "" { - botIntent.SendNotice(ctx, evt.RoomID, fmt.Sprintf("Topic '%s' is already bridged to room %s.", topicName, existing)) - return - } - - spaceRoomID, err := h.DB.GetSpaceRoomID(string(evt.Sender)) - if err != nil { - log.Printf("[matrix] db error checking space: %v", err) - botIntent.SendNotice(ctx, evt.RoomID, "Internal database error.") - return - } - - if spaceRoomID == "" { - resp, err := botIntent.Client.CreateRoom(ctx, &mautrix.ReqCreateRoom{ - Name: "Notifications", - Preset: "private_chat", - CreationContent: map[string]interface{}{ - "type": "m.space", - }, - Invite: []id.UserID{evt.Sender}, - }) - if err != nil { - botIntent.SendNotice(ctx, evt.RoomID, fmt.Sprintf("Failed to create Notifications space: %v", err)) - return - } - spaceRoomID = string(resp.RoomID) - - if err := h.DB.SetSpaceRoomID(string(evt.Sender), spaceRoomID); err != nil { - log.Printf("[matrix] failed to save space: %v", err) - botIntent.SendNotice(ctx, evt.RoomID, "Failed to save space reference to database.") - return - } - - userIntent := h.AS.Intent(evt.Sender) - userIntent.IsCustomPuppet = true - if err := userIntent.EnsureJoined(ctx, id.RoomID(spaceRoomID)); err != nil { - log.Printf("[matrix] double puppet join space failed: %v", err) - } - - botIntent.SendNotice(ctx, evt.RoomID, "Created 'Notifications' space.") - } - - roomResp, err := botIntent.Client.CreateRoom(ctx, &mautrix.ReqCreateRoom{ - Name: topicName, - Topic: fmt.Sprintf("ntfy topic: %s", topicName), - Preset: "private_chat", - Invite: []id.UserID{evt.Sender}, - }) - if err != nil { - botIntent.SendNotice(ctx, evt.RoomID, fmt.Sprintf("Failed to create bridged room: %v", err)) - return - } - bridgeRoomID := roomResp.RoomID - - viaServers := []string{h.HSDomain} - if _, err := botIntent.SendStateEvent(ctx, id.RoomID(spaceRoomID), event.StateSpaceChild, string(bridgeRoomID), &event.SpaceChildEventContent{ - Via: viaServers, - }); err != nil { - log.Printf("[matrix] failed to add room to space: %v", err) - } - - userIntent := h.AS.Intent(evt.Sender) - userIntent.IsCustomPuppet = true - if err := userIntent.EnsureJoined(ctx, bridgeRoomID); err != nil { - log.Printf("[matrix] double puppet join bridge room failed: %v", err) - } - - if err := h.DB.SetBridge(topicName, string(bridgeRoomID)); err != nil { - log.Printf("[matrix] failed to save bridge: %v", err) - botIntent.SendNotice(ctx, evt.RoomID, "Bridge created but failed to save to database.") - return - } - - h.Ntfy.StartNtfyListener(topicName, string(bridgeRoomID), h.AS.BotIntent(), h.AS.BotMXID()) - - botIntent.SendNotice(ctx, evt.RoomID, fmt.Sprintf( - "Bridge created!\n• Topic: %s\n• Room: %s\nMessages sent in that room will be published to ntfy, and ntfy messages will appear there.", - topicName, bridgeRoomID, - )) -} - -func (h *MatrixHandler) handleMembership(ctx context.Context, evt *event.Event) { - member := evt.Content.AsMember() - if member == nil { - return - } - - stateKey := evt.GetStateKey() - if stateKey == "" { - return - } - - if member.Membership == event.MembershipInvite && id.UserID(stateKey) == h.AS.BotMXID() { - botIntent := h.AS.BotIntent() - if err := botIntent.EnsureJoined(ctx, evt.RoomID); err != nil { - log.Printf("[matrix] failed to auto-join room %s: %v", evt.RoomID, err) - } - } -} diff --git a/ntfy_client.go b/ntfy.go similarity index 52% rename from ntfy_client.go rename to ntfy.go index 1850354..cdb597d 100644 --- a/ntfy_client.go +++ b/ntfy.go @@ -16,6 +16,15 @@ import ( "maunium.net/go/mautrix/id" ) +const ( + defaultNtfyURL = "https://ntfy.sh" + connectTimeout = 10 * time.Second + initialBackoff = 1 * time.Second + maxBackoff = 2 * time.Minute + scannerBufSize = 64 * 1024 + maxScannerBufSize = 1024 * 1024 +) + type NtfyEvent struct { ID string `json:"id"` Time int64 `json:"time"` @@ -26,15 +35,24 @@ type NtfyEvent struct { } type NtfyClient struct { - BaseURL string + baseURL string + httpClient *http.Client } func NewNtfyClient(baseURL string) *NtfyClient { - return &NtfyClient{BaseURL: strings.TrimRight(baseURL, "/")} + if baseURL == "" { + baseURL = defaultNtfyURL + } + return &NtfyClient{ + baseURL: strings.TrimRight(baseURL, "/"), + httpClient: &http.Client{ + Timeout: connectTimeout, + }, + } } func (c *NtfyClient) PublishToNtfy(topic, message string) error { - url := fmt.Sprintf("%s/%s", c.BaseURL, topic) + url := fmt.Sprintf("%s/%s", c.baseURL, topic) req, err := http.NewRequest("POST", url, bytes.NewReader([]byte(message))) if err != nil { @@ -42,7 +60,7 @@ func (c *NtfyClient) PublishToNtfy(topic, message string) error { } req.Header.Set("Content-Type", "text/plain") - resp, err := http.DefaultClient.Do(req) + resp, err := c.httpClient.Do(req) if err != nil { return fmt.Errorf("publish request: %w", err) } @@ -55,33 +73,32 @@ func (c *NtfyClient) PublishToNtfy(topic, message string) error { return nil } -func (c *NtfyClient) StartNtfyListener( - topic, - matrixRoomID string, - intent *appservice.IntentAPI, - botUserID id.UserID, -) { - go c.listenLoop(topic, matrixRoomID, intent, botUserID) -} - -func (c *NtfyClient) listenLoop( - topic, - matrixRoomID string, - intent *appservice.IntentAPI, - botUserID id.UserID, -) { - backoff := 1 * time.Second - const maxBackoff = 2 * time.Minute +func (c *NtfyClient) Listen(ctx context.Context, topic, matrixRoomID string, intent *appservice.IntentAPI) { + backoff := initialBackoff for { - err := c.stream(topic, matrixRoomID, intent, botUserID) - if err != nil { - log.Printf("[ntfy] stream for topic %q disconnected: %v, reconnecting in %v", topic, err, backoff) - } else { - log.Printf("[ntfy] stream for topic %q closed cleanly, reconnecting in %v", topic, backoff) - backoff = 1 * time.Second + err := c.stream(ctx, topic, matrixRoomID, intent) + + select { + case <-ctx.Done(): + log.Printf("[ntfy] listener for topic %q stopped", topic) + return + default: } - time.Sleep(backoff) + + if err != nil { + log.Printf("[ntfy] topic %q disconnected: %v, reconnecting in %v", topic, err, backoff) + } else { + log.Printf("[ntfy] topic %q stream closed, reconnecting in %v", topic, backoff) + backoff = initialBackoff + } + + select { + case <-ctx.Done(): + return + case <-time.After(backoff): + } + backoff *= 2 if backoff > maxBackoff { backoff = maxBackoff @@ -89,16 +106,16 @@ func (c *NtfyClient) listenLoop( } } -func (c *NtfyClient) stream( - topic, - matrixRoomID string, - intent *appservice.IntentAPI, - botUserID id.UserID, -) error { - url := fmt.Sprintf("%s/%s/json", c.BaseURL, topic) +func (c *NtfyClient) stream(ctx context.Context, topic, matrixRoomID string, intent *appservice.IntentAPI) error { + url := fmt.Sprintf("%s/%s/json", c.baseURL, topic) - client := &http.Client{Timeout: 0} - resp, err := client.Get(url) + req, err := http.NewRequestWithContext(ctx, "GET", url, nil) + if err != nil { + return fmt.Errorf("create request: %w", err) + } + + streamClient := &http.Client{Timeout: 0} + resp, err := streamClient.Do(req) if err != nil { return fmt.Errorf("GET %s: %w", url, err) } @@ -108,10 +125,8 @@ func (c *NtfyClient) stream( return fmt.Errorf("GET %s returned %d", url, resp.StatusCode) } - ctx := context.Background() scanner := bufio.NewScanner(resp.Body) - buf := make([]byte, 0, 64*1024) - scanner.Buffer(buf, 1024*1024) + scanner.Buffer(make([]byte, 0, scannerBufSize), maxScannerBufSize) for scanner.Scan() { line := scanner.Bytes() @@ -123,36 +138,22 @@ func (c *NtfyClient) stream( if err := json.Unmarshal(line, &ntfyEvt); err != nil { continue } - - if ntfyEvt.Event != "message" { + if ntfyEvt.Event != "message" || ntfyEvt.Message == "" { continue } body := ntfyEvt.Message - if body == "" { - continue - } if ntfyEvt.Title != "" { body = fmt.Sprintf("**%s**\n%s", ntfyEvt.Title, body) } if _, err := intent.SendText(ctx, id.RoomID(matrixRoomID), body); err != nil { - log.Printf("[ntfy] failed to send to Matrix room %s: %v", matrixRoomID, err) + log.Printf("[ntfy] send to room %s failed: %v", matrixRoomID, err) } } if err := scanner.Err(); err != nil { - return fmt.Errorf("scanner error: %w", err) + return fmt.Errorf("scanner: %w", err) } return nil } - -func (c *NtfyClient) IsBridgingBack(sender id.UserID, botUserID id.UserID, ghostPrefix string) bool { - if sender == botUserID { - return true - } - if ghostPrefix != "" && strings.HasPrefix(string(sender), ghostPrefix) { - return true - } - return false -}