pico

created pr with 103.1 on 2026-01-05T14:52:02Z · by c8ef7d19

added 103.2 on 2026-01-08T01:19:34Z · by c8ef7d19


                      1:  66cf9b9 = 1:  66cf9b9 feat(pipe): add pipe_monitors table


                      2:  83b0c70 = 2:  83b0c70 chore(pipe): add new db methods


                      3:  43aa4a1 = 3:  43aa4a1 chore(pipe): add db impl


                      4:  a75186b = 4:  a75186b chore(pipe): add tests for monitoring


                      5:  9c7fcda = 5:  9c7fcda feat(pipe): add monitor cli


                      6:  2b86dec = 6:  2b86dec feat(pipe): status and rss commands


                      7:  8df2aa2 = 7:  8df2aa2 feat(pipe): monitor pub and pipe cmd


                      8:  11a5bce = 8:  11a5bce chore(pipe): monitor help text


                      9:  670350c = 9:  670350c refactor(pipe): monitor pipes on throttle interval


                      10:  ade2b83 = 10:  ade2b83 fix: window and ping fixes


                      11:  68f5cd1 = 11:  68f5cd1 chore(pipe): add tests


                      -:  ------- > 12:  74fb40f chore(pipe): add logging stmts


                      -:  ------- > 13:  c422818 refactor(pipe): create pipe monitors history table


                      -:  ------- > 14:  e0c3a88 fix(pipe): only update monitor if within window


                      -:  ------- > 15:  9a8fbde chore(pipe): implement monitor history db interface


                      -:  ------- > 16:  f168507 feat(pipe): record historical monitors


                      -:  ------- > 17:  16df77f feat(pipe): monitor calculate uptime


                      -:  ------- > 18:  10ffdf1 feat(pipe): uptime cli command


                      -:  ------- > 19:  8231ac5 chore(pipe): monitor history stubs


                      -:  ------- > 20:  65a63a9 fix(pipe): uptime cmd

added 103.3 on 2026-02-01T17:16:24Z · by c8ef7d19


                      3:  43aa4a1 ! 1:  603ca6e chore(pubsub): add more tests


                      1:  66cf9b9 < -:  ------- feat(pipe): add pipe_monitors table


                      2:  83b0c70 < -:  ------- chore(pipe): add new db methods


                      4:  a75186b ! 2:  17e00b2 feat(pubsub): round robin


                      13:  c422818 ! 3:  e3136bd fix(pubsub): check for eof before processing and skip empty byte reads


                      -:  ------- > 4:  9a6d19e fix: rr


                      5:  9c7fcda < -:  ------- feat(pipe): add monitor cli


                      16:  f168507 ! 5:  5b3f3a1 fix: sending 0 byte read


                      6:  2b86dec < -:  ------- feat(pipe): status and rss commands


                      11:  68f5cd1 ! 6:  4fff471 refactor: fixes


                      7:  8df2aa2 < -:  ------- feat(pipe): monitor pub and pipe cmd


                      14:  e0c3a88 ! 7:  d0dfc85 chore: SetDispatch on Broker


                      8:  11a5bce < -:  ------- chore(pipe): monitor help text


                      9:  670350c < -:  ------- refactor(pipe): monitor pipes on throttle interval


                      10:  ade2b83 < -:  ------- fix: window and ping fixes


                      12:  74fb40f < -:  ------- chore(pipe): add logging stmts


                      15:  9a8fbde < -:  ------- chore(pipe): implement monitor history db interface


                      17:  16df77f < -:  ------- feat(pipe): monitor calculate uptime


                      18:  10ffdf1 < -:  ------- feat(pipe): uptime cli command


                      19:  8231ac5 < -:  ------- chore(pipe): monitor history stubs


                      20:  65a63a9 < -:  ------- fix(pipe): uptime cmd

cmds

checkout latest patchset:

ssh pr.pico.sh print 103 | git am -3

checkout any patchset in a patch request:

ssh pr.pico.sh print 103.[rev] | git am -3

add changes to patch request:

git format-patch main --stdout | ssh pr.pico.sh pr add 103

Patchset `103.1` on 2026-01-05T14:52:02Z · commit 2b86dec

feat(pipe): add pipe_monitors table

Eric Bower 2025-12-26T21:02:39Z

chore(pipe): add new db methods

Eric Bower 2025-12-26T21:24:18Z

chore(pipe): add db impl

Eric Bower 2025-12-26T21:37:49Z

chore(pipe): add tests for monitoring

Eric Bower 2025-12-28T14:41:40Z

feat(pipe): add monitor cli

Eric Bower 2025-12-28T14:44:04Z

→ feat(pipe): status and rss commands

Eric Bower 2025-12-28T14:55:12Z

feat(pipe): monitor pub and pipe cmd

Eric Bower 2025-12-28T15:26:40Z

chore(pipe): monitor help text

Eric Bower 2025-12-28T15:40:57Z

refactor(pipe): monitor pipes on throttle interval

Eric Bower 2025-12-28T16:22:23Z

fix: window and ping fixes

Eric Bower 2026-01-03T02:26:40Z

chore(pipe): add tests

Eric Bower 2026-01-03T03:50:20Z

feat(pipe): status and rss commands

Eric Bower 2025-12-28T14:55:12Z

Semantic diff summary

2 added, 3 modified, 0 signature changed, 0 removed across 2 analyzed files

pkg/apps/pipe/cli.go

chunk lines 9-20 modified
function_declaration Middleware modified
method_declaration status added
method_declaration rss added

pkg/db/db.go

chunk lines 391-402 modified

+110 -0 pkg/apps/pipe/cli.go #

......@@ -9,10 +9,12 @@ import (
	"log/slog"
	"slices"
	"strings"
	"text/tabwriter"
	"time"

	"github.com/antoniomika/syncmap"
	"github.com/google/uuid"
	"github.com/gorilla/feeds"
	"github.com/picosh/pico/pkg/db"
	"github.com/picosh/pico/pkg/pssh"
	"github.com/picosh/pico/pkg/shared"

......@@ -77,6 +79,18 @@ func Middleware(handler *CliHandler) pssh.SSHServerMiddleware {
					sesh.Fatal(err)
				}
				return next(sesh)
			case "status":
				err := handler.status(cliCmd, user)
				if err != nil {
					sesh.Fatal(err)
				}
				return next(sesh)
			case "rss":
				err := handler.rss(cliCmd, user)
				if err != nil {
					sesh.Fatal(err)
				}
				return next(sesh)
			}

			topic := ""

......@@ -343,6 +357,102 @@ func (handler *CliHandler) monitor(cmd *CliCmd, user *db.User) error {
	return nil
}

func (handler *CliHandler) status(cmd *CliCmd, user *db.User) error {
	if user == nil {
		return fmt.Errorf("access denied")
	}

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

	if len(monitors) == 0 {
		_, _ = fmt.Fprintln(cmd.sesh, "no monitors found")
		return nil
	}

	writer := tabwriter.NewWriter(cmd.sesh, 0, 0, 2, ' ', tabwriter.TabIndent)
	_, _ = fmt.Fprintln(writer, "Topic\tStatus\tReason\tWindow\tLast Ping\tCreated")

	for _, m := range monitors {
		status := "healthy"
		reason := ""
		if err := m.Status(); err != nil {
			status = "unhealthy"
			reason = err.Error()
		}

		lastPing := "never"
		if m.LastPing != nil {
			lastPing = m.LastPing.Format("2006-01-02 15:04:05")
		}

		createdAt := ""
		if m.CreatedAt != nil {
			createdAt = m.CreatedAt.Format("2006-01-02 15:04:05")
		}

		_, _ = fmt.Fprintf(
			writer,
			"%s\t%s\t%s\t%s\t%s\t%s\r\n",
			m.Topic,
			status,
			reason,
			m.WindowDur.String(),
			lastPing,
			createdAt,
		)
	}
	_ = writer.Flush()
	return nil
}

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

	monitors, err := handler.DBPool.FindPipeMonitorsByUser(user.ID)
	if err != nil {
		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)},
		Description: "Alerts for pipe monitor status changes",
		Author:      &feeds.Author{Name: user.Name},
		Created:     now,
	}

	var feedItems []*feeds.Item
	for _, m := range monitors {
		if err := m.Status(); err != nil {
			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)},
				Description: err.Error(),
				Created:     now,
				Updated:     now,
				Author:      &feeds.Author{Name: user.Name},
			}
			feedItems = append(feedItems, item)
		}
	}
	feed.Items = feedItems

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

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

func (handler *CliHandler) pub(cmd *CliCmd, topic string, clientID string) error {
	pubCmd := flagSet("pub", cmd.sesh)
	access := pubCmd.String("a", "", "Comma separated list of pico usernames or ssh-key fingerprints to allow access to a topic")

+6 -0 pkg/db/db.go #

......@@ -391,6 +391,12 @@ type PipeMonitor struct {
}

func (m *PipeMonitor) Status() error {
	if m.LastPing == nil {
		return fmt.Errorf("no ping received yet")
	}
	if m.WindowEnd == nil {
		return fmt.Errorf("window end not set")
	}
	windowStart := m.WindowEnd.Add(-m.WindowDur)
	lastPingAfterStart := m.LastPing.After(windowStart)
	if !lastPingAfterStart {

...	...		@@ -343,6 +357,102 @@ func (handler CliHandler) monitor(cmd CliCmd, user *db.User) error {
343	357		return nil
344	358		}
345	359
	360	+	func (handler CliHandler) status(cmd CliCmd, user *db.User) error {
	361	+	if user == nil {
	362	+	return fmt.Errorf("access denied")
	363	+	}
	364	+
	365	+	monitors, err := handler.DBPool.FindPipeMonitorsByUser(user.ID)
	366	+	if err != nil {
	367	+	return fmt.Errorf("failed to fetch monitors: %w", err)
	368	+	}
	369	+
	370	+	if len(monitors) == 0 {
	371	+	_, _ = fmt.Fprintln(cmd.sesh, "no monitors found")
	372	+	return nil
	373	+	}
	374	+
	375	+	writer := tabwriter.NewWriter(cmd.sesh, 0, 0, 2, ' ', tabwriter.TabIndent)
	376	+	_, _ = fmt.Fprintln(writer, "Topic\tStatus\tReason\tWindow\tLast Ping\tCreated")
	377	+
	378	+	for _, m := range monitors {
	379	+	status := "healthy"
	380	+	reason := ""
	381	+	if err := m.Status(); err != nil {
	382	+	status = "unhealthy"
	383	+	reason = err.Error()
	384	+	}
	385	+
	386	+	lastPing := "never"
	387	+	if m.LastPing != nil {
	388	+	lastPing = m.LastPing.Format("2006-01-02 15:04:05")
	389	+	}
	390	+
	391	+	createdAt := ""
	392	+	if m.CreatedAt != nil {
	393	+	createdAt = m.CreatedAt.Format("2006-01-02 15:04:05")
	394	+	}
	395	+
	396	+	_, _ = fmt.Fprintf(
	397	+	writer,
	398	+	"%s\t%s\t%s\t%s\t%s\t%s\r\n",
	399	+	m.Topic,
	400	+	status,
	401	+	reason,
	402	+	m.WindowDur.String(),
	403	+	lastPing,
	404	+	createdAt,
	405	+	)
	406	+	}
	407	+	_ = writer.Flush()
	408	+	return nil
	409	+	}
	410	+
	411	+	func (handler CliHandler) rss(cmd CliCmd, user *db.User) error {
	412	+	if user == nil {
	413	+	return fmt.Errorf("access denied")
	414	+	}
	415	+
	416	+	monitors, err := handler.DBPool.FindPipeMonitorsByUser(user.ID)
	417	+	if err != nil {
	418	+	return fmt.Errorf("failed to fetch monitors: %w", err)
	419	+	}
	420	+
	421	+	now := time.Now()
	422	+	feed := &feeds.Feed{
	423	+	Title: fmt.Sprintf("Pipe Monitors for %s", user.Name),
	424	+	Link: &feeds.Link{Href: fmt.Sprintf("https://%s", handler.Cfg.Domain)},
	425	+	Description: "Alerts for pipe monitor status changes",
	426	+	Author: &feeds.Author{Name: user.Name},
	427	+	Created: now,
	428	+	}
	429	+
	430	+	var feedItems []*feeds.Item
	431	+	for _, m := range monitors {
	432	+	if err := m.Status(); err != nil {
	433	+	item := &feeds.Item{
	434	+	Id: fmt.Sprintf("%s-%s-%d", user.ID, m.Topic, now.Unix()),
	435	+	Title: fmt.Sprintf("ALERT: %s is unhealthy", m.Topic),
	436	+	Link: &feeds.Link{Href: fmt.Sprintf("https://%s", handler.Cfg.Domain)},
	437	+	Description: err.Error(),
	438	+	Created: now,
	439	+	Updated: now,
	440	+	Author: &feeds.Author{Name: user.Name},
	441	+	}
	442	+	feedItems = append(feedItems, item)
	443	+	}
	444	+	}
	445	+	feed.Items = feedItems
	446	+
	447	+	rss, err := feed.ToRss()
	448	+	if err != nil {
	449	+	return fmt.Errorf("failed to generate RSS: %w", err)
	450	+	}
	451	+
	452	+	_, _ = fmt.Fprint(cmd.sesh, rss)
	453	+	return nil
	454	+	}
	455	+
346	456		func (handler CliHandler) pub(cmd CliCmd, topic string, clientID string) error {
347	457		pubCmd := flagSet("pub", cmd.sesh)
348	458		access := pubCmd.String("a", "", "Comma separated list of pico usernames or ssh-key fingerprints to allow access to a topic")

...	...		@@ -391,6 +391,12 @@ type PipeMonitor struct {
391	391		}
392	392
393	393		func (m *PipeMonitor) Status() error {
	394	+	if m.LastPing == nil {
	395	+	return fmt.Errorf("no ping received yet")
	396	+	}
	397	+	if m.WindowEnd == nil {
	398	+	return fmt.Errorf("window end not set")
	399	+	}
394	400		windowStart := m.WindowEnd.Add(-m.WindowDur)
395	401		lastPingAfterStart := m.LastPing.After(windowStart)
396	402		if !lastPingAfterStart {

...	...		@@ -9,10 +9,12 @@ import (
9	9		"log/slog"
10	10		"slices"
11	11		"strings"
	12	+	"text/tabwriter"
12	13		"time"
13	14
14	15		"github.com/antoniomika/syncmap"
15	16		"github.com/google/uuid"
	17	+	"github.com/gorilla/feeds"
16	18		"github.com/picosh/pico/pkg/db"
17	19		"github.com/picosh/pico/pkg/pssh"
18	20		"github.com/picosh/pico/pkg/shared"

...	...		@@ -77,6 +79,18 @@ func Middleware(handler *CliHandler) pssh.SSHServerMiddleware {
77	79		sesh.Fatal(err)
78	80		}
79	81		return next(sesh)
	82	+	case "status":
	83	+	err := handler.status(cliCmd, user)
	84	+	if err != nil {
	85	+	sesh.Fatal(err)
	86	+	}
	87	+	return next(sesh)
	88	+	case "rss":
	89	+	err := handler.rss(cliCmd, user)
	90	+	if err != nil {
	91	+	sesh.Fatal(err)
	92	+	}
	93	+	return next(sesh)
80	94		}
81	95
82	96		topic := ""

pico

Patchset 103.1 on 2026-01-05T14:52:02Z · commit 2b86dec

Patchset `103.1` on 2026-01-05T14:52:02Z · commit 2b86dec