feat(notification): Telegram add support for topics in groups (#894)

* feat(notification): send Telegram messages to a specific topic of a group

* Convert settings.Topic to integer once and reuse it as part of the
telegramSender struct.

* feat(notifications): add migrations for topic

* fix(notifications): find null string

* fix(notifications): form initial values

---------

Co-authored-by: ze0s <ze0s@riseup.net>

internal/database/notification.go

@@ -31,7 +31,7 @@ func NewNotificationRepo(log logger.Logger, db *DB) domain.NotificationRepo {
 func (r *NotificationRepo) Find(ctx context.Context, params domain.NotificationQueryParams) ([]domain.Notification, int, error) {
 
 	queryBuilder := r.db.squirrel.
-		Select("id", "name", "type", "enabled", "events", "webhook", "token", "api_key", "channel", "priority", "created_at", "updated_at", "COUNT(*) OVER() AS total_count").
+		Select("id", "name", "type", "enabled", "events", "webhook", "token", "api_key", "channel", "priority", "topic", "created_at", "updated_at", "COUNT(*) OVER() AS total_count").
 		From("notification").
 		OrderBy("name")
 
@@ -52,9 +52,9 @@ func (r *NotificationRepo) Find(ctx context.Context, params domain.NotificationQ
 	for rows.Next() {
 		var n domain.Notification
 
-		var webhook, token, apiKey, channel sql.NullString
+		var webhook, token, apiKey, channel, topic sql.NullString
 
-		if err := rows.Scan(&n.ID, &n.Name, &n.Type, &n.Enabled, pq.Array(&n.Events), &webhook, &token, &apiKey, &channel, &n.Priority, &n.CreatedAt, &n.UpdatedAt, &totalCount); err != nil {
+		if err := rows.Scan(&n.ID, &n.Name, &n.Type, &n.Enabled, pq.Array(&n.Events), &webhook, &token, &apiKey, &channel, &n.Priority, &topic, &n.CreatedAt, &n.UpdatedAt, &totalCount); err != nil {
 			return nil, 0, errors.Wrap(err, "error scanning row")
 		}
 
@@ -62,6 +62,7 @@ func (r *NotificationRepo) Find(ctx context.Context, params domain.NotificationQ
 		n.Webhook = webhook.String
 		n.Token = token.String
 		n.Channel = channel.String
+		n.Topic = topic.String
 
 		notifications = append(notifications, n)
 	}
@@ -74,7 +75,7 @@ func (r *NotificationRepo) Find(ctx context.Context, params domain.NotificationQ
 
 func (r *NotificationRepo) List(ctx context.Context) ([]domain.Notification, error) {
 
-	rows, err := r.db.handler.QueryContext(ctx, "SELECT id, name, type, enabled, events, token, api_key,  webhook, title, icon, host, username, password, channel, targets, devices, priority,created_at, updated_at FROM notification ORDER BY name ASC")
+	rows, err := r.db.handler.QueryContext(ctx, "SELECT id, name, type, enabled, events, token, api_key,  webhook, title, icon, host, username, password, channel, targets, devices, priority, topic, created_at, updated_at FROM notification ORDER BY name ASC")
 	if err != nil {
 		return nil, errors.Wrap(err, "error executing query")
 	}
@@ -86,8 +87,8 @@ func (r *NotificationRepo) List(ctx context.Context) ([]domain.Notification, err
 		var n domain.Notification
 		//var eventsSlice []string
 
-		var token, apiKey, webhook, title, icon, host, username, password, channel, targets, devices sql.NullString
+		var token, apiKey, webhook, title, icon, host, username, password, channel, targets, devices, topic sql.NullString
-		if err := rows.Scan(&n.ID, &n.Name, &n.Type, &n.Enabled, pq.Array(&n.Events), &token, &apiKey, &webhook, &title, &icon, &host, &username, &password, &channel, &targets, &devices, &n.Priority, &n.CreatedAt, &n.UpdatedAt); err != nil {
+		if err := rows.Scan(&n.ID, &n.Name, &n.Type, &n.Enabled, pq.Array(&n.Events), &token, &apiKey, &webhook, &title, &icon, &host, &username, &password, &channel, &targets, &devices, &n.Priority, &topic, &n.CreatedAt, &n.UpdatedAt); err != nil {
 			return nil, errors.Wrap(err, "error scanning row")
 		}
 
@@ -103,6 +104,7 @@ func (r *NotificationRepo) List(ctx context.Context) ([]domain.Notification, err
 		n.Channel = channel.String
 		n.Targets = targets.String
 		n.Devices = devices.String
+		n.Topic = topic.String
 
 		notifications = append(notifications, n)
 	}
@@ -134,6 +136,7 @@ func (r *NotificationRepo) FindByID(ctx context.Context, id int) (*domain.Notifi
 			"targets",
 			"devices",
 			"priority",
+			"topic",
 			"created_at",
 			"updated_at",
 		).
@@ -152,8 +155,8 @@ func (r *NotificationRepo) FindByID(ctx context.Context, id int) (*domain.Notifi
 
 	var n domain.Notification
 
-	var token, apiKey, webhook, title, icon, host, username, password, channel, targets, devices sql.NullString
+	var token, apiKey, webhook, title, icon, host, username, password, channel, targets, devices, topic sql.NullString
-	if err := row.Scan(&n.ID, &n.Name, &n.Type, &n.Enabled, pq.Array(&n.Events), &token, &apiKey, &webhook, &title, &icon, &host, &username, &password, &channel, &targets, &devices, &n.Priority, &n.CreatedAt, &n.UpdatedAt); err != nil {
+	if err := row.Scan(&n.ID, &n.Name, &n.Type, &n.Enabled, pq.Array(&n.Events), &token, &apiKey, &webhook, &title, &icon, &host, &username, &password, &channel, &targets, &devices, &n.Priority, &topic, &n.CreatedAt, &n.UpdatedAt); err != nil {
 		return nil, errors.Wrap(err, "error scanning row")
 	}
 
@@ -168,6 +171,7 @@ func (r *NotificationRepo) FindByID(ctx context.Context, id int) (*domain.Notifi
 	n.Channel = channel.String
 	n.Targets = targets.String
 	n.Devices = devices.String
+	n.Topic = topic.String
 
 	return &n, nil
 }
@@ -177,6 +181,7 @@ func (r *NotificationRepo) Store(ctx context.Context, notification domain.Notifi
 	token := toNullString(notification.Token)
 	apiKey := toNullString(notification.APIKey)
 	channel := toNullString(notification.Channel)
+	topic := toNullString(notification.Topic)
 
 	queryBuilder := r.db.squirrel.
 		Insert("notification").
@@ -190,6 +195,7 @@ func (r *NotificationRepo) Store(ctx context.Context, notification domain.Notifi
 			"api_key",
 			"channel",
 			"priority",
+			"topic",
 		).
 		Values(
 			notification.Name,
@@ -201,6 +207,7 @@ func (r *NotificationRepo) Store(ctx context.Context, notification domain.Notifi
 			apiKey,
 			channel,
 			notification.Priority,
+			topic,
 		).
 		Suffix("RETURNING id").RunWith(r.db.handler)
 
@@ -222,6 +229,7 @@ func (r *NotificationRepo) Update(ctx context.Context, notification domain.Notif
 	token := toNullString(notification.Token)
 	apiKey := toNullString(notification.APIKey)
 	channel := toNullString(notification.Channel)
+	topic := toNullString(notification.Topic)
 
 	queryBuilder := r.db.squirrel.
 		Update("notification").
@@ -234,6 +242,7 @@ func (r *NotificationRepo) Update(ctx context.Context, notification domain.Notif
 		Set("api_key", apiKey).
 		Set("channel", channel).
 		Set("priority", notification.Priority).
+		Set("topic", topic).
 		Set("updated_at", sq.Expr("CURRENT_TIMESTAMP")).
 		Where(sq.Eq{"id": notification.ID})
 

internal/database/postgres_migrate.go

@@ -303,6 +303,7 @@ CREATE TABLE notification
 	rooms      TEXT,
 	targets    TEXT,
 	devices    TEXT,
+	topic      TEXT,
 	priority   INTEGER DEFAULT 0,
 	created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
 	updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
@@ -677,4 +678,6 @@ ADD COLUMN download_url TEXT;
 	`,
 	`ALTER TABLE notification
 ADD COLUMN priority INTEGER DEFAULT 0;`,
+	`ALTER TABLE notification
+ADD COLUMN topic text;`,
 }

internal/database/sqlite_migrate.go

@@ -295,6 +295,7 @@ CREATE TABLE notification
 	rooms      TEXT,
 	targets    TEXT,
 	devices    TEXT,
+	topic      TEXT,
 	priority   INTEGER DEFAULT 0,
 	created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
 	updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
@@ -1070,4 +1071,6 @@ ADD COLUMN download_url TEXT;
 	`,
 	`ALTER TABLE notification
 ADD COLUMN priority INTEGER DEFAULT 0;`,
+	`ALTER TABLE notification
+ADD COLUMN topic text;`,
 }

internal/domain/notification.go

@@ -41,6 +41,7 @@ type Notification struct {
 	Targets   string           `json:"targets"`
 	Devices   string           `json:"devices"`
 	Priority  int32            `json:"priority"`
+	Topic     string           `json:"topic"`
 	CreatedAt time.Time        `json:"created_at"`
 	UpdatedAt time.Time        `json:"updated_at"`
 }

internal/notification/telegram.go

@@ -10,6 +10,7 @@ import (
 	"html"
 	"io"
 	"net/http"
+	"strconv"
 	"strings"
 	"time"
 
@@ -19,29 +20,42 @@ import (
 	"github.com/rs/zerolog"
 )
 
+// Reference: https://core.telegram.org/bots/api#sendmessage
 type TelegramMessage struct {
-	ChatID    string `json:"chat_id"`
+	ChatID          string `json:"chat_id"`
-	Text      string `json:"text"`
+	Text            string `json:"text"`
-	ParseMode string `json:"parse_mode"`
+	ParseMode       string `json:"parse_mode"`
+	MessageThreadID int    `json:"message_thread_id,omitempty"`
 }
 
 type telegramSender struct {
 	log      zerolog.Logger
 	Settings domain.Notification
+	ThreadID int
 }
 
 func NewTelegramSender(log zerolog.Logger, settings domain.Notification) domain.NotificationSender {
+	threadID := 0
+	if t := settings.Topic; t != "" {
+		var err error
+		threadID, err = strconv.Atoi(t)
+		if err != nil {
+			log.Error().Err(err).Msgf("could not parse specified topic %q as an integer", t)
+		}
+	}
 	return &telegramSender{
 		log:      log.With().Str("sender", "telegram").Logger(),
 		Settings: settings,
+		ThreadID: threadID,
 	}
 }
 
 func (s *telegramSender) Send(event domain.NotificationEvent, payload domain.NotificationPayload) error {
 	m := TelegramMessage{
-		ChatID:    s.Settings.Channel,
+		ChatID:          s.Settings.Channel,
-		Text:      s.buildMessage(event, payload),
+		Text:            s.buildMessage(event, payload),
-		ParseMode: "HTML",
+		MessageThreadID: s.ThreadID,
+		ParseMode:       "HTML",
 		//ParseMode: "MarkdownV2",
 	}
 

web/src/forms/settings/NotificationForms.tsx

@@ -121,6 +121,11 @@ function FormFieldsTelegram() {
         label="Chat ID"
         help="Chat ID"
       />
+      <PasswordFieldWide
+        name="topic"
+        label="Message Thread ID"
+        help="Message Thread (topic) of a Supergroup"
+      />
     </div>
   );
 }
@@ -436,6 +441,7 @@ interface InitialValues {
   api_key?: string;
   priority?: number;
   channel?: string;
+  topic?: string;
   events: NotificationEvent[];
 }
 
@@ -484,6 +490,7 @@ export function NotificationUpdateForm({ isOpen, toggle, notification }: UpdateP
     api_key: notification.api_key,
     priority: notification.priority,
     channel: notification.channel,
+    topic: notification.topic,
     events: notification.events || []
   };
 
@@ -567,4 +574,4 @@ export function NotificationUpdateForm({ isOpen, toggle, notification }: UpdateP
       )}
     </SlideOver>
   );
-}
+}

web/src/types/Notification.d.ts

@@ -4,7 +4,13 @@
  */
 
 type NotificationType = "DISCORD" | "NOTIFIARR" | "TELEGRAM" | "PUSHOVER";
-type NotificationEvent = "PUSH_APPROVED" | "PUSH_REJECTED" | "PUSH_ERROR" | "IRC_DISCONNECTED" | "IRC_RECONNECTED" | "APP_UPDATE_AVAILABLE";
+type NotificationEvent =
+  "PUSH_APPROVED"
+  | "PUSH_REJECTED"
+  | "PUSH_ERROR"
+  | "IRC_DISCONNECTED"
+  | "IRC_RECONNECTED"
+  | "APP_UPDATE_AVAILABLE";
 
 interface Notification {
   id: number;
@@ -17,4 +23,5 @@ interface Notification {
   api_key?: string;
   channel?: string;
   priority?: number;
+  topic?: string;
 }