lazyntfy/internal/ntfy/manager.go

package ntfy

import (
	"context"
	"errors"
	"strconv"
	"strings"
	"sync"
	"time"

	"lazyntfy/internal/config"
)

// Notice is a stream lifecycle update or a decoded event.
type Notice struct {
	Event  *Event
	Topics []string
	Status string
	Up     bool
	UpSet  bool
}

// Manager keeps one reconnecting JSON stream per topic group.
type Manager struct {
	client     *Client
	out        chan Notice
	reload     chan struct{}
	mu         sync.Mutex
	ctx        context.Context
	subs       []config.Sub
	since      string
	cursors    map[string]int64
	backoff    time.Duration
	maxBackoff time.Duration
}

// NewManager starts with no subscriptions until Start is called.
func NewManager(client *Client) *Manager {
	if client == nil {
		client = NewClient()
	}
	return &Manager{
		client:     client,
		out:        make(chan Notice, 256),
		reload:     make(chan struct{}, 1),
		cursors:    map[string]int64{},
		backoff:    time.Second,
		maxBackoff: 30 * time.Second,
	}
}

// SetBackoff overrides reconnect delays. Tests use a short interval.
func (m *Manager) SetBackoff(initial, max time.Duration) {
	m.backoff = initial
	m.maxBackoff = max
}

// Prime records already-seen newest times per origin/topic so the first stream
// does not replay history the client restored from disk. Keys use the same
// origin + "/" + topic format the manager tracks internally.
func (m *Manager) Prime(cursors map[string]int64) {
	m.mu.Lock()
	defer m.mu.Unlock()
	for key, ts := range cursors {
		if ts > m.cursors[key] {
			m.cursors[key] = ts
		}
	}
}

// Notices is the stream of events and status lines. It is never closed.
func (m *Manager) Notices() <-chan Notice { return m.out }

// Start subscribes until ctx is cancelled. Call it once.
func (m *Manager) Start(ctx context.Context, subs []config.Sub, since string) {
	m.mu.Lock()
	m.ctx = ctx
	m.subs = append([]config.Sub(nil), subs...)
	m.since = since
	m.mu.Unlock()
	go m.run()
}

// Reload drops the current connections and subscribes again.
func (m *Manager) Reload() {
	select {
	case m.reload <- struct{}{}:
	default:
	}
}

// Add appends a topic for this session and reconnects.
func (m *Manager) Add(sub config.Sub) {
	m.mu.Lock()
	m.subs = append(m.subs, sub)
	m.mu.Unlock()
	m.Reload()
}

func (m *Manager) run() {
	ctx := m.ctx
	for {
		if ctx.Err() != nil {
			return
		}
		m.mu.Lock()
		subs := append([]config.Sub(nil), m.subs...)
		m.mu.Unlock()
		child, cancel := context.WithCancel(ctx)
		var wg sync.WaitGroup
		if len(subs) == 0 {
			m.emit(child, Notice{Status: "no topics"})
		}
		for _, group := range GroupSubs(subs) {
			wg.Add(1)
			go func(g Group) {
				defer wg.Done()
				m.supervise(child, g)
			}(group)
		}
		stopped := make(chan struct{})
		go func() {
			wg.Wait()
			close(stopped)
		}()
		select {
		case <-ctx.Done():
			cancel()
			<-stopped
			return
		case <-m.reload:
			cancel()
			<-stopped
		}
	}
}

func (m *Manager) supervise(ctx context.Context, g Group) {
	delay := m.backoff
	for {
		if ctx.Err() != nil {
			return
		}
		err := m.consume(ctx, g)
		if ctx.Err() != nil || errors.Is(err, context.Canceled) {
			return
		}
		status := "disconnected, retrying"
		if err != nil {
			status = err.Error()
		}
		m.emit(ctx, Notice{Topics: g.Names, Status: status, Up: false, UpSet: true})
		timer := time.NewTimer(delay)
		select {
		case <-ctx.Done():
			timer.Stop()
			return
		case <-timer.C:
		}
		if delay < m.maxBackoff {
			delay *= 2
			if delay > m.maxBackoff {
				delay = m.maxBackoff
			}
		}
	}
}

func (m *Manager) consume(ctx context.Context, g Group) error {
	resp, meta, err := m.client.OpenStream(ctx, g.URL(m.sinceFor(g)), g.Auth)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	m.emit(ctx, Notice{Topics: g.Names, Status: "connected", Up: true, UpSet: true})
	if meta.Truncated {
		m.emit(ctx, Notice{Topics: g.Names, Status: "cache truncated to the newest messages", Up: true, UpSet: true})
	}
	return ReadLines(resp.Body, func(ev Event) error {
		if ctx.Err() != nil {
			return ctx.Err()
		}
		m.note(g.Origin, ev)
		copyEv := ev
		m.emit(ctx, Notice{Event: &copyEv, Topics: []string{ev.Topic}})
		return nil
	}, func(err error) {
		m.emit(ctx, Notice{Status: "skipped malformed event"})
	})
}

func (m *Manager) sinceFor(g Group) string {
	m.mu.Lock()
	defer m.mu.Unlock()
	var max int64
	for _, name := range g.Names {
		if ts := m.cursors[g.Origin+"/"+name]; ts > max {
			max = ts
		}
	}
	if max == 0 {
		return m.since
	}
	return strconv.FormatInt(max, 10)
}

func (m *Manager) note(origin string, ev Event) {
	if ev.Time == 0 || ev.Topic == "" || strings.Contains(ev.Topic, ",") {
		return
	}
	key := origin + "/" + ev.Topic
	m.mu.Lock()
	if ev.Time > m.cursors[key] {
		m.cursors[key] = ev.Time
	}
	m.mu.Unlock()
}

func (m *Manager) emit(ctx context.Context, n Notice) {
	select {
	case <-ctx.Done():
	case m.out <- n:
	}
}