ProjectOctopus/message_pusher/pusher/client.go

package pusher

import (
	"bytes"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"regexp"
	"strings"
	"wdd.io/agent-common/logger"
)

var (
	topicRegex          = regexp.MustCompile(`^[-_A-Za-z0-9]{1,64}$`) // Same as in server/server.go
	log                 = logger.Log
	DefaultPusherClient = NewDefaultClient()
)

const (
	maxResponseBytes = 4096
)

type Client struct {
	config *Config
}

// Message is a struct that represents a ntfy message
type Message struct { // TODO combine with server.message
	ID         string
	Event      string
	Time       int64
	Topic      string
	Message    string
	Title      string
	Priority   int
	Tags       []string
	Click      string
	Icon       string
	Attachment *Attachment

	// Additional fields
	TopicURL       string
	SubscriptionID string
	Raw            string
}

// Attachment represents a message attachment
type Attachment struct {
	Name    string `json:"name"`
	Type    string `json:"type,omitempty"`
	Size    int64  `json:"size,omitempty"`
	Expires int64  `json:"expires,omitempty"`
	URL     string `json:"url"`
	Owner   string `json:"-"` // IP address of uploader, used for rate limiting
}

// New creates a new Client using a given Config
func New(config *Config) *Client {
	return &Client{
		config: config,
	}
}

func NewDefaultClient() *Client {
	defaultConfig := NewDefaultConfig()
	return New(defaultConfig)
}

func (c *Client) ChangeTopicName(topicName string) {
	c.config.DefaultTopic = topicName
}

func (c *Client) PublishDefault(message bytes.Buffer, options []PublishOption) (*Message, error) {
	if c.config.DefaultTopic == "" {
		return nil, errors.New("[PublishDefault] - topic empty")
	}
	// parse default
	options = c.parseConfigToOption(options)

	return c.PublishReader(c.config.DefaultTopic, bytes.NewReader(message.Bytes()), options)
}

// Publish sends a message to a specific topic, optionally using options.
// See PublishReader for details.
func (c *Client) Publish(topic, message string, options []PublishOption) (*Message, error) {
	return c.PublishReader(topic, strings.NewReader(message), options)
}

// PublishReader sends a message to a specific topic, optionally using options.
//
// A topic can be either a full URL (e.g. https://myhost.lan/mytopic), a short URL which is then prepended https://
// (e.g. myhost.lan -> https://myhost.lan), or a short name which is expanded using the default host in the
// config (e.g. mytopic -> https://ntfy.sh/mytopic).
//
// To pass title, priority and tags, check out WithTitle, WithPriority, WithTagsList, WithDelay, WithNoCache,
// WithNoFirebase, and the generic WithHeader.
func (c *Client) PublishReader(topic string, body io.Reader, options []PublishOption) (*Message, error) {
	topicURL, err := c.expandTopicURL(topic)
	if err != nil {
		return nil, err
	}
	req, err := http.NewRequest("POST", topicURL, body)
	if err != nil {
		return nil, err
	}
	for _, option := range options {
		if err := option(req); err != nil {
			return nil, err
		}
	}
	log.DebugF("%s Publishing message with headers %s", topicURL, req.Header)
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	b, err := io.ReadAll(io.LimitReader(resp.Body, maxResponseBytes))
	if err != nil {
		return nil, err
	}
	if resp.StatusCode != http.StatusOK {
		return nil, errors.New(strings.TrimSpace(string(b)))
	}
	m, err := toMessage(string(b), topicURL, "")
	if err != nil {
		return nil, err
	}
	return m, nil
}

func (c *Client) expandTopicURL(topic string) (string, error) {
	if strings.HasPrefix(topic, "http://") || strings.HasPrefix(topic, "https://") {
		return topic, nil
	} else if strings.Contains(topic, "/") {
		return fmt.Sprintf("https://%s", topic), nil
	}
	if !topicRegex.MatchString(topic) {
		return "", fmt.Errorf("invalid topic name: %s", topic)
	}
	return fmt.Sprintf("%s/%s", c.config.Host, topic), nil
}

func (c *Client) parseConfigToOption(options []PublishOption) []PublishOption {
	config := c.config

	if config.Token != "" {
		options = append(options, WithBearerAuth(config.Token))
	} else if config.User != "" {
		if *config.Password != "" {
			options = append(options, WithBasicAuth(config.User, *config.Password))
		} else {
			log.ErrorF("[parseConfigToOption] - default password is empty!")
		}
	}

	return options
}

func toMessage(s, topicURL, subscriptionID string) (*Message, error) {
	var m *Message
	if err := json.NewDecoder(strings.NewReader(s)).Decode(&m); err != nil {
		return nil, err
	}
	m.TopicURL = topicURL
	m.SubscriptionID = subscriptionID
	m.Raw = s
	return m, nil
}