initial commit

ntfy_client.go

@@ -0,0 +1,158 @@
+package main
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"log"
+	"net/http"
+	"strings"
+	"time"
+
+	"maunium.net/go/mautrix/appservice"
+	"maunium.net/go/mautrix/id"
+)
+
+type NtfyEvent struct {
+	ID      string `json:"id"`
+	Time    int64  `json:"time"`
+	Event   string `json:"event"`
+	Topic   string `json:"topic"`
+	Message string `json:"message"`
+	Title   string `json:"title"`
+}
+
+type NtfyClient struct {
+	BaseURL string
+}
+
+func NewNtfyClient(baseURL string) *NtfyClient {
+	return &NtfyClient{BaseURL: strings.TrimRight(baseURL, "/")}
+}
+
+func (c *NtfyClient) PublishToNtfy(topic, message string) error {
+	url := fmt.Sprintf("%s/%s", c.BaseURL, topic)
+
+	req, err := http.NewRequest("POST", url, bytes.NewReader([]byte(message)))
+	if err != nil {
+		return fmt.Errorf("create request: %w", err)
+	}
+	req.Header.Set("Content-Type", "text/plain")
+
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return fmt.Errorf("publish request: %w", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusCreated && resp.StatusCode != http.StatusOK {
+		body, _ := io.ReadAll(resp.Body)
+		return fmt.Errorf("ntfy returned %d: %s", resp.StatusCode, string(body))
+	}
+	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
+
+	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
+		}
+		time.Sleep(backoff)
+		backoff *= 2
+		if backoff > maxBackoff {
+			backoff = maxBackoff
+		}
+	}
+}
+
+func (c *NtfyClient) stream(
+	topic,
+	matrixRoomID string,
+	intent *appservice.IntentAPI,
+	botUserID id.UserID,
+) error {
+	url := fmt.Sprintf("%s/%s/json", c.BaseURL, topic)
+
+	client := &http.Client{Timeout: 0}
+	resp, err := client.Get(url)
+	if err != nil {
+		return fmt.Errorf("GET %s: %w", url, err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		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)
+
+	for scanner.Scan() {
+		line := scanner.Bytes()
+		if len(line) == 0 {
+			continue
+		}
+
+		var ntfyEvt NtfyEvent
+		if err := json.Unmarshal(line, &ntfyEvt); err != nil {
+			continue
+		}
+
+		if ntfyEvt.Event != "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)
+		}
+	}
+
+	if err := scanner.Err(); err != nil {
+		return fmt.Errorf("scanner error: %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
+}