add telegram and ntfy event delivery

2026-07-26 20:31:42 +04:00
parent f818003ad5
commit 4093bdb683
3 changed files with 147 additions and 0 deletions
@@ -0,0 +1,129 @@
+package delivery
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"orchestra/internal/domain"
+	"time"
+)
+
+type Sender interface {
+	Send(context.Context, string) error
+}
+
+type Telegram struct {
+	Token, ChatID string
+	Client        *http.Client
+}
+
+func (t Telegram) Send(ctx context.Context, message string) error {
+	if t.Token == "" || t.ChatID == "" {
+		return fmt.Errorf("telegram credentials missing")
+	}
+	b, _ := json.Marshal(map[string]string{"chat_id": t.ChatID, "text": message})
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, "https://api.telegram.org/bot"+t.Token+"/sendMessage", bytes.NewReader(b))
+	if err != nil {
+		return err
+	}
+	req.Header.Set("Content-Type", "application/json")
+	c := t.Client
+	if c == nil {
+		c = http.DefaultClient
+	}
+	resp, err := c.Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode/100 != 2 {
+		return fmt.Errorf("telegram: %s", resp.Status)
+	}
+	return nil
+}
+
+type Ntfy struct {
+	Topic, Token string
+	Client       *http.Client
+	URL          string
+}
+
+func (n Ntfy) Send(ctx context.Context, message string) error {
+	if n.Topic == "" {
+		return fmt.Errorf("ntfy topic missing")
+	}
+	u := n.URL
+	if u == "" {
+		u = "https://ntfy.sh/" + n.Topic
+	} else {
+		u += "/" + n.Topic
+	}
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, u, bytes.NewBufferString(message))
+	if err != nil {
+		return err
+	}
+	if n.Token != "" {
+		req.Header.Set("Authorization", "Bearer "+n.Token)
+	}
+	c := n.Client
+	if c == nil {
+		c = http.DefaultClient
+	}
+	resp, err := c.Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode/100 != 2 {
+		return fmt.Errorf("ntfy: %s", resp.Status)
+	}
+	return nil
+}
+
+func Message(e domain.Event) (string, bool) {
+	switch e.Type {
+	case "TaskCompleted":
+		return "Task completed: " + e.TaskID, true
+	case "TaskFailed":
+		return "Task failed: " + e.TaskID, true
+	case "TaskBlocked":
+		return "Task blocked: " + e.TaskID, true
+	case "ApprovalRequested":
+		return "Approval requested: " + e.TaskID, true
+	default:
+		return "", false
+	}
+}
+
+type Fanout struct {
+	Senders []Sender
+	Cursor  uint64
+	Retry   time.Duration
+}
+
+func (f *Fanout) Run(ctx context.Context, events func(uint64) []domain.Event) error {
+	if f.Retry <= 0 {
+		f.Retry = 5 * time.Second
+	}
+	for {
+		for _, e := range events(f.Cursor) {
+			if msg, ok := Message(e); ok {
+				for _, s := range f.Senders {
+					if err := s.Send(ctx, msg); err != nil {
+						return err
+					}
+				}
+			}
+			if e.Seq > f.Cursor {
+				f.Cursor = e.Seq
+			}
+		}
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-time.After(f.Retry):
+		}
+	}
+}