HODA/internal/peer/peer.go

// Package peer lets the primary pull app lists and icons from agents on other
// Docker hosts, authenticating with the shared PEER_TOKEN.
package peer

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"net/http"
	"net/url"
	"strings"
	"sync"
	"time"

	"git.mariosem.es/msemes/hoda/internal/apps"
	"git.mariosem.es/msemes/hoda/internal/icons"
)

const (
	requestTimeout = 3 * time.Second
	maxBodyBytes   = 4 << 20
	maxIconBytes   = 1 << 20
)

// Puller fetches each agent's apps on an interval and its icons on demand.
type Puller struct {
	peers    []string
	token    string
	interval time.Duration
	store    *apps.Store
	log      *slog.Logger
	client   *http.Client

	mu    sync.Mutex
	names map[string]string     // agent URL -> host name it last reported
	up    map[string]bool       // agent URL -> reachable on the last pull
	icons map[string]icons.Icon // app ID -> icon fetched from its agent
}

// New returns a Puller for the agents at peers.
func New(peers []string, token string, interval time.Duration, store *apps.Store, log *slog.Logger) *Puller {
	return &Puller{
		peers:    peers,
		token:    token,
		interval: interval,
		store:    store,
		log:      log,
		client:   &http.Client{Timeout: requestTimeout},
		names:    map[string]string{},
		up:       map[string]bool{},
		icons:    map[string]icons.Icon{},
	}
}

// Run pulls every agent now and then once per interval until ctx ends.
func (p *Puller) Run(ctx context.Context) {
	ticker := time.NewTicker(p.interval)
	defer ticker.Stop()
	for {
		var wg sync.WaitGroup
		for _, peer := range p.peers {
			wg.Add(1)
			go func() {
				defer wg.Done()
				p.pull(ctx, peer)
			}()
		}
		wg.Wait()
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
		}
	}
}

func (p *Puller) pull(ctx context.Context, peer string) {
	var body apps.Response
	err := p.get(ctx, peer+"/api/apps", func(resp *http.Response) error {
		return json.NewDecoder(io.LimitReader(resp.Body, maxBodyBytes)).Decode(&body)
	})
	if err == nil && len(body.Hosts) == 0 {
		err = errors.New("the agent reported no host")
	}
	if err != nil {
		if ctx.Err() != nil {
			return
		}
		p.store.SetError(peer, p.name(peer))
		p.mark(peer, false, err)
		return
	}

	status := body.Hosts[0]
	list := make([]apps.App, 0, len(body.Apps))
	for _, a := range body.Apps {
		// The agent holds the token, but a bad link from it would still end up in an href.
		if !apps.ValidID(a.ID) || !apps.IsWebURL(a.URL) {
			continue
		}
		if a.InternalURL != "" && !apps.IsWebURL(a.InternalURL) {
			a.InternalURL = ""
		}
		if !strings.HasPrefix(a.Icon, "/icons/"+a.ID+"?") {
			a.Icon = ""
		}
		if a.Host == "" {
			a.Host = status.Name
		}
		list = append(list, a)
	}
	issues := make([]apps.Issue, 0, len(body.Issues))
	for _, issue := range body.Issues {
		if issue.Host == "" {
			issue.Host = status.Name
		}
		issues = append(issues, issue)
	}
	p.mu.Lock()
	p.names[peer] = status.Name
	p.mu.Unlock()
	p.store.SetReported(peer, status, list, issues)
	p.mark(peer, true, nil)
	p.pruneIcons()
}

// Icon returns an agent's icon for one of its apps, fetched once per icon version.
func (p *Puller) Icon(ctx context.Context, id string) (icons.Icon, bool) {
	key, app, ok := p.store.Lookup(id)
	if !ok || key == apps.LocalSource || app.Icon == "" {
		return icons.Icon{}, false
	}
	version := ""
	if u, err := url.Parse(app.Icon); err == nil {
		version = u.Query().Get("v")
	}
	p.mu.Lock()
	cached, hit := p.icons[id]
	p.mu.Unlock()
	if hit && cached.Version == version {
		return cached, true
	}

	var ic icons.Icon
	err := p.get(ctx, key+"/icons/"+id, func(resp *http.Response) error {
		data, err := io.ReadAll(io.LimitReader(resp.Body, maxIconBytes))
		if err != nil {
			return err
		}
		ctype := resp.Header.Get("Content-Type")
		if !strings.HasPrefix(ctype, "image/") {
			return fmt.Errorf("agent sent %q, not an image", ctype)
		}
		ic = icons.Icon{Data: data, ContentType: ctype, Version: version}
		return nil
	})
	if err != nil {
		p.log.Debug("fetching an icon from an agent failed", "agent", key, "app", app.Name, "err", err)
		return icons.Icon{}, false
	}
	p.mu.Lock()
	p.icons[id] = ic
	p.mu.Unlock()
	return ic, true
}

func (p *Puller) get(ctx context.Context, link string, read func(*http.Response) error) error {
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, link, nil)
	if err != nil {
		return err
	}
	req.Header.Set("Authorization", "Bearer "+p.token)
	resp, err := p.client.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		return fmt.Errorf("%s: %s", link, resp.Status)
	}
	return read(resp)
}

// name is the host name an agent last reported, or its URL's host before that.
func (p *Puller) name(peer string) string {
	p.mu.Lock()
	defer p.mu.Unlock()
	if n := p.names[peer]; n != "" {
		return n
	}
	if u, err := url.Parse(peer); err == nil && u.Hostname() != "" {
		return u.Hostname()
	}
	return peer
}

// mark logs only when an agent goes up or down, not on every pull.
func (p *Puller) mark(peer string, up bool, err error) {
	p.mu.Lock()
	was, seen := p.up[peer]
	p.up[peer] = up
	p.mu.Unlock()
	switch {
	case up && (!seen || !was):
		p.log.Info("agent reachable", "agent", peer)
	case !up && (!seen || was):
		p.log.Warn("can't reach agent, keeping its last known apps", "agent", peer, "err", err)
	}
}

func (p *Puller) pruneIcons() {
	p.mu.Lock()
	ids := make([]string, 0, len(p.icons))
	for id := range p.icons {
		ids = append(ids, id)
	}
	p.mu.Unlock()
	for _, id := range ids {
		if _, _, ok := p.store.Lookup(id); !ok {
			p.mu.Lock()
			delete(p.icons, id)
			p.mu.Unlock()
		}
	}
}