orchestra/internal/delivery/delivery.go

package delivery

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"log"
	"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
	// OnError is called for each failed send instead of aborting the fanout
	// goroutine (S4, AUDIT.md: a single ntfy hiccup used to `return` and
	// permanently kill notifications for the rest of the process). Defaults
	// to log.Printf.
	OnError func(err error)
	// SaveCursor, if set, is called after the cursor advances past a
	// processed event so a restart can resume from here instead of
	// re-notifying the entire log from seq 0 (S4).
	SaveCursor func(cursor uint64)
}

func (f *Fanout) Run(ctx context.Context, events func(uint64) []domain.Event) error {
	if f.Retry <= 0 {
		f.Retry = 5 * time.Second
	}
	onError := f.OnError
	if onError == nil {
		onError = func(err error) { log.Printf("delivery: %v", err) }
	}
	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 {
						onError(fmt.Errorf("send %s to %T: %w", msg, s, err))
					}
				}
			}
			if e.Seq > f.Cursor {
				f.Cursor = e.Seq
				if f.SaveCursor != nil {
					f.SaveCursor(f.Cursor)
				}
			}
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(f.Retry):
		}
	}
}