159 lines
3.4 KiB
Go
159 lines
3.4 KiB
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"
|
|
)
|
|
|
|
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
|
|
}
|