mautrix-ntfy/ntfy.go

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"
)

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"`
	Event   string `json:"event"`
	Topic   string `json:"topic"`
	Message string `json:"message"`
	Title   string `json:"title"`
}

type NtfyClient struct {
	baseURL    string
	httpClient *http.Client
}

func NewNtfyClient(baseURL string) *NtfyClient {
	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)

	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 := c.httpClient.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) Listen(ctx context.Context, topic, matrixRoomID string, intent *appservice.IntentAPI) {
	backoff := initialBackoff

	for {
		err := c.stream(ctx, topic, matrixRoomID, intent)

		select {
		case <-ctx.Done():
			log.Printf("[ntfy] listener for topic %q stopped", topic)
			return
		default:
		}

		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
		}
	}
}

func (c *NtfyClient) stream(ctx context.Context, topic, matrixRoomID string, intent *appservice.IntentAPI) error {
	url := fmt.Sprintf("%s/%s/json", c.baseURL, topic)

	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)
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		return fmt.Errorf("GET %s returned %d", url, resp.StatusCode)
	}

	scanner := bufio.NewScanner(resp.Body)
	scanner.Buffer(make([]byte, 0, scannerBufSize), maxScannerBufSize)

	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" || ntfyEvt.Message == "" {
			continue
		}

		body := ntfyEvt.Message
		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] send to room %s failed: %v", matrixRoomID, err)
		}
	}

	if err := scanner.Err(); err != nil {
		return fmt.Errorf("scanner: %w", err)
	}
	return nil
}