feat(pipe): web based rss feed

Commit `bb7b4f2`

Eric Bower · 2026-01-08 22:13:26 -0500 EST

parent b2aa9b3

feat(pipe): web based rss feed

2 files changed, +51, -12

M pkg/apps/pipe/api.go

M pkg/apps/pipe/cli.go

M pkg/apps/pipe/api.go

+41, -2

......@@ -16,6 +16,7 @@ import (

	"github.com/google/uuid"
	"github.com/gorilla/websocket"
	"github.com/picosh/pico/pkg/db"
	"github.com/picosh/pico/pkg/db/postgres"
	"github.com/picosh/pico/pkg/shared"
	"github.com/picosh/utils/pipe"

......@@ -386,10 +387,48 @@ func handlePipe() http.HandlerFunc {
	}
}

func createMainRoutes(staticRoutes []shared.Route) []shared.Route {
func rssHandler(cfg *shared.ConfigSite, dbpool db.DB) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		apiToken, _ := url.PathUnescape(shared.GetField(r, 0))
		user, err := dbpool.FindUserByToken(apiToken)
		if err != nil {
			cfg.Logger.Error(
				"could not find user for token",
				"err", err.Error(),
				"token", apiToken,
			)
			http.Error(w, "invalid token", http.StatusNotFound)
			return
		}
		rss, err := MonitorRss(dbpool, user, cfg.Domain)
		if err != nil {
			cfg.Logger.Error(
				"error generating monitor rss feed",
				"err", err,
				"token", apiToken,
			)
			http.Error(w, "error generating monitor rss feed", http.StatusInternalServerError)
			return
		}

		_, err = w.Write([]byte(rss))
		if err != nil {
			cfg.Logger.Error(
				"error with rss response writer",
				"err", err,
				"token", apiToken,
			)
			http.Error(w, "error generating monitor rss feederror with rss response writer", http.StatusInternalServerError)
			return
		}
	}
}

func createMainRoutes(staticRoutes []shared.Route, cfg *shared.ConfigSite, dbpool db.DB) []shared.Route {
	routes := []shared.Route{
		shared.NewRoute("GET", "/", shared.CreatePageHandler("html/marketing.page.tmpl")),
		shared.NewRoute("GET", "/check", shared.CheckHandler),
		shared.NewRoute("GET", "/rss/(.+)", rssHandler(cfg, dbpool)),
		shared.NewRoute("GET", "/_metrics", promhttp.Handler().ServeHTTP),
	}

......@@ -428,7 +467,7 @@ func StartApiServer() {
		staticRoutes = shared.CreatePProfRoutes(staticRoutes)
	}

	mainRoutes := createMainRoutes(staticRoutes)
	mainRoutes := createMainRoutes(staticRoutes, cfg, db)
	subdomainRoutes := staticRoutes

	info := shared.NewPicoPipeClient()

M pkg/apps/pipe/cli.go

+10, -10

......@@ -93,7 +93,8 @@ func Middleware(handler *CliHandler) pssh.SSHServerMiddleware {
				}
				return next(sesh)
			case "rss":
				err := handler.rss(cliCmd, user)
				rss, err := MonitorRss(handler.DBPool, user, handler.Cfg.Domain)
				_, _ = fmt.Fprintln(sesh, rss)
				if err != nil {
					logger.Error("rss cmd", "err", err)
					sesh.Fatal(err)

......@@ -542,20 +543,20 @@ func parseDuration(s string) (time.Duration, error) {
	return time.ParseDuration(s)
}

func (handler *CliHandler) rss(cmd *CliCmd, user *db.User) error {
func MonitorRss(dbpool db.DB, user *db.User, domain string) (string, error) {
	if user == nil {
		return fmt.Errorf("access denied")
		return "", fmt.Errorf("access denied")
	}

	monitors, err := handler.DBPool.FindPipeMonitorsByUser(user.ID)
	monitors, err := dbpool.FindPipeMonitorsByUser(user.ID)
	if err != nil {
		return fmt.Errorf("failed to fetch monitors: %w", err)
		return "", fmt.Errorf("failed to fetch monitors: %w", err)
	}

	now := time.Now()
	feed := &feeds.Feed{
		Title:       fmt.Sprintf("Pipe Monitors for %s", user.Name),
		Link:        &feeds.Link{Href: fmt.Sprintf("https://%s", handler.Cfg.Domain)},
		Link:        &feeds.Link{Href: fmt.Sprintf("https://%s", domain)},
		Description: "Alerts for pipe monitor status changes",
		Author:      &feeds.Author{Name: user.Name},
		Created:     now,

......@@ -567,7 +568,7 @@ func (handler *CliHandler) rss(cmd *CliCmd, user *db.User) error {
			item := &feeds.Item{
				Id:          fmt.Sprintf("%s-%s-%d", user.ID, m.Topic, now.Unix()),
				Title:       fmt.Sprintf("ALERT: %s is unhealthy", m.Topic),
				Link:        &feeds.Link{Href: fmt.Sprintf("https://%s", handler.Cfg.Domain)},
				Link:        &feeds.Link{Href: fmt.Sprintf("https://%s", domain)},
				Description: err.Error(),
				Created:     now,
				Updated:     now,

......@@ -580,11 +581,10 @@ func (handler *CliHandler) rss(cmd *CliCmd, user *db.User) error {

	rss, err := feed.ToRss()
	if err != nil {
		return fmt.Errorf("failed to generate RSS: %w", err)
		return "", fmt.Errorf("failed to generate RSS: %w", err)
	}

	_, _ = fmt.Fprint(cmd.sesh, rss)
	return nil
	return rss, nil
}

func (handler *CliHandler) pub(cmd *CliCmd, topic string, clientID string) error {

...	...		@@ -386,10 +387,48 @@ func handlePipe() http.HandlerFunc {
386	387		}
387	388		}
388	389
389		-	func createMainRoutes(staticRoutes []shared.Route) []shared.Route {
	390	+	func rssHandler(cfg *shared.ConfigSite, dbpool db.DB) http.HandlerFunc {
	391	+	return func(w http.ResponseWriter, r *http.Request) {
	392	+	apiToken, _ := url.PathUnescape(shared.GetField(r, 0))
	393	+	user, err := dbpool.FindUserByToken(apiToken)
	394	+	if err != nil {
	395	+	cfg.Logger.Error(
	396	+	"could not find user for token",
	397	+	"err", err.Error(),
	398	+	"token", apiToken,
	399	+	)
	400	+	http.Error(w, "invalid token", http.StatusNotFound)
	401	+	return
	402	+	}
	403	+	rss, err := MonitorRss(dbpool, user, cfg.Domain)
	404	+	if err != nil {
	405	+	cfg.Logger.Error(
	406	+	"error generating monitor rss feed",
	407	+	"err", err,
	408	+	"token", apiToken,
	409	+	)
	410	+	http.Error(w, "error generating monitor rss feed", http.StatusInternalServerError)
	411	+	return
	412	+	}
	413	+
	414	+	_, err = w.Write([]byte(rss))
	415	+	if err != nil {
	416	+	cfg.Logger.Error(
	417	+	"error with rss response writer",
	418	+	"err", err,
	419	+	"token", apiToken,
	420	+	)
	421	+	http.Error(w, "error generating monitor rss feederror with rss response writer", http.StatusInternalServerError)
	422	+	return
	423	+	}
	424	+	}
	425	+	}
	426	+
	427	+	func createMainRoutes(staticRoutes []shared.Route, cfg *shared.ConfigSite, dbpool db.DB) []shared.Route {
390	428		routes := []shared.Route{
391	429		shared.NewRoute("GET", "/", shared.CreatePageHandler("html/marketing.page.tmpl")),
392	430		shared.NewRoute("GET", "/check", shared.CheckHandler),
	431	+	shared.NewRoute("GET", "/rss/(.+)", rssHandler(cfg, dbpool)),
393	432		shared.NewRoute("GET", "/_metrics", promhttp.Handler().ServeHTTP),
394	433		}
395	434

...	...		@@ -428,7 +467,7 @@ func StartApiServer() {
428	467		staticRoutes = shared.CreatePProfRoutes(staticRoutes)
429	468		}
430	469
431		-	mainRoutes := createMainRoutes(staticRoutes)
	470	+	mainRoutes := createMainRoutes(staticRoutes, cfg, db)
432	471		subdomainRoutes := staticRoutes
433	472
434	473		info := shared.NewPicoPipeClient()

...	...		@@ -542,20 +543,20 @@ func parseDuration(s string) (time.Duration, error) {
542	543		return time.ParseDuration(s)
543	544		}
544	545
545		-	func (handler CliHandler) rss(cmd CliCmd, user *db.User) error {
	546	+	func MonitorRss(dbpool db.DB, user *db.User, domain string) (string, error) {
546	547		if user == nil {
547		-	return fmt.Errorf("access denied")
	548	+	return "", fmt.Errorf("access denied")
548	549		}
549	550
550		-	monitors, err := handler.DBPool.FindPipeMonitorsByUser(user.ID)
	551	+	monitors, err := dbpool.FindPipeMonitorsByUser(user.ID)
551	552		if err != nil {
552		-	return fmt.Errorf("failed to fetch monitors: %w", err)
	553	+	return "", fmt.Errorf("failed to fetch monitors: %w", err)
553	554		}
554	555
555	556		now := time.Now()
556	557		feed := &feeds.Feed{
557	558		Title: fmt.Sprintf("Pipe Monitors for %s", user.Name),
558		-	Link: &feeds.Link{Href: fmt.Sprintf("https://%s", handler.Cfg.Domain)},
	559	+	Link: &feeds.Link{Href: fmt.Sprintf("https://%s", domain)},
559	560		Description: "Alerts for pipe monitor status changes",
560	561		Author: &feeds.Author{Name: user.Name},
561	562		Created: now,

...	...		@@ -580,11 +581,10 @@ func (handler CliHandler) rss(cmd CliCmd, user *db.User) error {
580	581
581	582		rss, err := feed.ToRss()
582	583		if err != nil {
583		-	return fmt.Errorf("failed to generate RSS: %w", err)
	584	+	return "", fmt.Errorf("failed to generate RSS: %w", err)
584	585		}
585	586
586		-	_, _ = fmt.Fprint(cmd.sesh, rss)
587		-	return nil
	587	+	return rss, nil
588	588		}
589	589
590	590		func (handler CliHandler) pub(cmd CliCmd, topic string, clientID string) error {

...	...		@@ -16,6 +16,7 @@ import (
16	16
17	17		"github.com/google/uuid"
18	18		"github.com/gorilla/websocket"
	19	+	"github.com/picosh/pico/pkg/db"
19	20		"github.com/picosh/pico/pkg/db/postgres"
20	21		"github.com/picosh/pico/pkg/shared"
21	22		"github.com/picosh/utils/pipe"

...	...		@@ -93,7 +93,8 @@ func Middleware(handler *CliHandler) pssh.SSHServerMiddleware {
93	93		}
94	94		return next(sesh)
95	95		case "rss":
96		-	err := handler.rss(cliCmd, user)
	96	+	rss, err := MonitorRss(handler.DBPool, user, handler.Cfg.Domain)
	97	+	_, _ = fmt.Fprintln(sesh, rss)
97	98		if err != nil {
98	99		logger.Error("rss cmd", "err", err)
99	100		sesh.Fatal(err)

...	...		@@ -567,7 +568,7 @@ func (handler CliHandler) rss(cmd CliCmd, user *db.User) error {
567	568		item := &feeds.Item{
568	569		Id: fmt.Sprintf("%s-%s-%d", user.ID, m.Topic, now.Unix()),
569	570		Title: fmt.Sprintf("ALERT: %s is unhealthy", m.Topic),
570		-	Link: &feeds.Link{Href: fmt.Sprintf("https://%s", handler.Cfg.Domain)},
	571	+	Link: &feeds.Link{Href: fmt.Sprintf("https://%s", domain)},
571	572		Description: err.Error(),
572	573		Created: now,
573	574		Updated: now,

Commit bb7b4f2

Commit `bb7b4f2`