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.2` on 2026-01-08T01:19:34Z · commit 74fb40f

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

→ chore(pipe): add logging stmts

Eric Bower 2026-01-05T14:46:42Z

refactor(pipe): create pipe monitors history table

Eric Bower 2026-01-07T20:00:47Z

fix(pipe): only update monitor if within window

Eric Bower 2026-01-07T21:52:08Z

chore(pipe): implement monitor history db interface

Eric Bower 2026-01-07T23:19:05Z

feat(pipe): record historical monitors

Eric Bower 2026-01-07T23:27:20Z

feat(pipe): monitor calculate uptime

Eric Bower 2026-01-07T23:27:20Z

feat(pipe): uptime cli command

Eric Bower 2026-01-08T00:48:50Z

chore(pipe): monitor history stubs

Eric Bower 2026-01-08T00:52:48Z

fix(pipe): uptime cmd

Eric Bower 2026-01-08T01:18:38Z

chore(pipe): add logging stmts

Eric Bower 2026-01-05T14:46:42Z

Semantic diff summary

0 added, 4 modified, 0 signature changed, 0 removed across 1 analyzed file

pkg/apps/pipe/cli.go

function_declaration Middleware modified
function_declaration monitor modified
function_declaration status modified
function_declaration updateMonitor modified

+20 -2 pkg/apps/pipe/cli.go #

......@@ -72,24 +72,28 @@ func Middleware(handler *CliHandler) pssh.SSHServerMiddleware {
			case "ls":
				err := handler.ls(cliCmd)
				if err != nil {
					logger.Error("ls cmd", "err", err)
					sesh.Fatal(err)
				}
				return next(sesh)
			case "monitor":
				err := handler.monitor(cliCmd, user)
				if err != nil {
					logger.Error("monitor cmd", "err", err)
					sesh.Fatal(err)
				}
				return next(sesh)
			case "status":
				err := handler.status(cliCmd, user)
				if err != nil {
					logger.Error("status cmd", "err", err)
					sesh.Fatal(err)
				}
				return next(sesh)
			case "rss":
				err := handler.rss(cliCmd, user)
				if err != nil {
					logger.Error("rss cmd", "err", err)
					sesh.Fatal(err)
				}
				return next(sesh)

......@@ -144,16 +148,19 @@ func Middleware(handler *CliHandler) pssh.SSHServerMiddleware {
			case "pub":
				err := handler.pub(cliCmd, topic, clientID)
				if err != nil {
					logger.Error("pub cmd", "err", err)
					sesh.Fatal(err)
				}
			case "sub":
				err := handler.sub(cliCmd, topic, clientID)
				if err != nil {
					logger.Error("sub cmd", "err", err)
					sesh.Fatal(err)
				}
			case "pipe":
				err := handler.pipe(cliCmd, topic, clientID)
				if err != nil {
					logger.Error("pipe cmd", "err", err)
					sesh.Fatal(err)
				}
			}

......@@ -341,6 +348,7 @@ func (handler *CliHandler) monitor(cmd *CliCmd, user *db.User) error {
	resolvedTopic := result.Name

	if *del {
		handler.Logger.Info("removing pipe monitor", "topic", resolvedTopic)
		err := handler.DBPool.RemovePipeMonitor(user.ID, resolvedTopic)
		if err != nil {
			return fmt.Errorf("failed to delete monitor: %w", err)

......@@ -368,6 +376,12 @@ func (handler *CliHandler) monitor(cmd *CliCmd, user *db.User) error {
	}

	winEnd := time.Now().UTC().Add(dur)
	handler.Logger.Info(
		"upserting pipe monitor",
		"topic", resolvedTopic,
		"dur", dur,
		"window", winEnd.UTC().Format(time.RFC3339),
	)
	err = handler.DBPool.UpsertPipeMonitor(user.ID, resolvedTopic, dur, &winEnd)
	if err != nil {
		return fmt.Errorf("failed to create monitor: %w", err)

......@@ -405,12 +419,12 @@ func (handler *CliHandler) status(cmd *CliCmd, user *db.User) error {

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

		windowEnd := ""
		if m.WindowEnd != nil {
			windowEnd = m.WindowEnd.UTC().Format("2006-01-02 15:04:05Z")
			windowEnd = m.WindowEnd.UTC().Format(time.RFC3339)
		}

		_, _ = fmt.Fprintf(

......@@ -704,8 +718,10 @@ func (handler *CliHandler) updateMonitor(cmd *CliCmd, topic string) {
		return
	}

	handler.Logger.Info("update monitor", "topic", topic)
	monitor, err := handler.DBPool.FindPipeMonitorByTopic(cmd.user.ID, topic)
	if err != nil || monitor == nil {
		handler.Logger.Info("no monitor found", "topic", topic)
		return
	}

......@@ -713,12 +729,14 @@ func (handler *CliHandler) updateMonitor(cmd *CliCmd, topic string) {
	if err := handler.DBPool.UpdatePipeMonitorLastPing(cmd.user.ID, topic, &now); err != nil {
		handler.Logger.Error("failed to update monitor last_ping", "err", err, "topic", topic)
	}
	handler.Logger.Info("update monitor ping", "topic", topic)

	// Always reset the window to now + duration on ping
	newWindowEnd := now.Add(monitor.WindowDur)
	if err := handler.DBPool.UpsertPipeMonitor(cmd.user.ID, topic, monitor.WindowDur, &newWindowEnd); err != nil {
		handler.Logger.Error("failed to reset monitor window", "err", err, "topic", topic)
	}
	handler.Logger.Info("updated monitor window", "topic", topic, "window", newWindowEnd.UTC().Format(time.RFC3339))
}

const monitorThrottleInterval = 15 * time.Second

...	...		@@ -144,16 +148,19 @@ func Middleware(handler *CliHandler) pssh.SSHServerMiddleware {
144	148		case "pub":
145	149		err := handler.pub(cliCmd, topic, clientID)
146	150		if err != nil {
	151	+	logger.Error("pub cmd", "err", err)
147	152		sesh.Fatal(err)
148	153		}
149	154		case "sub":
150	155		err := handler.sub(cliCmd, topic, clientID)
151	156		if err != nil {
	157	+	logger.Error("sub cmd", "err", err)
152	158		sesh.Fatal(err)
153	159		}
154	160		case "pipe":
155	161		err := handler.pipe(cliCmd, topic, clientID)
156	162		if err != nil {
	163	+	logger.Error("pipe cmd", "err", err)
157	164		sesh.Fatal(err)
158	165		}
159	166		}

...	...		@@ -368,6 +376,12 @@ func (handler CliHandler) monitor(cmd CliCmd, user *db.User) error {
368	376		}
369	377
370	378		winEnd := time.Now().UTC().Add(dur)
	379	+	handler.Logger.Info(
	380	+	"upserting pipe monitor",
	381	+	"topic", resolvedTopic,
	382	+	"dur", dur,
	383	+	"window", winEnd.UTC().Format(time.RFC3339),
	384	+	)
371	385		err = handler.DBPool.UpsertPipeMonitor(user.ID, resolvedTopic, dur, &winEnd)
372	386		if err != nil {
373	387		return fmt.Errorf("failed to create monitor: %w", err)

...	...		@@ -405,12 +419,12 @@ func (handler CliHandler) status(cmd CliCmd, user *db.User) error {
405	419
406	420		lastPing := "never"
407	421		if m.LastPing != nil {
408		-	lastPing = m.LastPing.UTC().Format("2006-01-02 15:04:05Z")
	422	+	lastPing = m.LastPing.UTC().Format(time.RFC3339)
409	423		}
410	424
411	425		windowEnd := ""
412	426		if m.WindowEnd != nil {
413		-	windowEnd = m.WindowEnd.UTC().Format("2006-01-02 15:04:05Z")
	427	+	windowEnd = m.WindowEnd.UTC().Format(time.RFC3339)
414	428		}
415	429
416	430		_, _ = fmt.Fprintf(

...	...		@@ -704,8 +718,10 @@ func (handler CliHandler) updateMonitor(cmd CliCmd, topic string) {
704	718		return
705	719		}
706	720
	721	+	handler.Logger.Info("update monitor", "topic", topic)
707	722		monitor, err := handler.DBPool.FindPipeMonitorByTopic(cmd.user.ID, topic)
708	723		if err != nil \|\| monitor == nil {
	724	+	handler.Logger.Info("no monitor found", "topic", topic)
709	725		return
710	726		}
711	727

...	...		@@ -713,12 +729,14 @@ func (handler CliHandler) updateMonitor(cmd CliCmd, topic string) {
713	729		if err := handler.DBPool.UpdatePipeMonitorLastPing(cmd.user.ID, topic, &now); err != nil {
714	730		handler.Logger.Error("failed to update monitor last_ping", "err", err, "topic", topic)
715	731		}
	732	+	handler.Logger.Info("update monitor ping", "topic", topic)
716	733
717	734		// Always reset the window to now + duration on ping
718	735		newWindowEnd := now.Add(monitor.WindowDur)
719	736		if err := handler.DBPool.UpsertPipeMonitor(cmd.user.ID, topic, monitor.WindowDur, &newWindowEnd); err != nil {
720	737		handler.Logger.Error("failed to reset monitor window", "err", err, "topic", topic)
721	738		}
	739	+	handler.Logger.Info("updated monitor window", "topic", topic, "window", newWindowEnd.UTC().Format(time.RFC3339))
722	740		}
723	741
724	742		const monitorThrottleInterval = 15 * time.Second

...	...		@@ -72,24 +72,28 @@ func Middleware(handler *CliHandler) pssh.SSHServerMiddleware {
72	72		case "ls":
73	73		err := handler.ls(cliCmd)
74	74		if err != nil {
	75	+	logger.Error("ls cmd", "err", err)
75	76		sesh.Fatal(err)
76	77		}
77	78		return next(sesh)
78	79		case "monitor":
79	80		err := handler.monitor(cliCmd, user)
80	81		if err != nil {
	82	+	logger.Error("monitor cmd", "err", err)
81	83		sesh.Fatal(err)
82	84		}
83	85		return next(sesh)
84	86		case "status":
85	87		err := handler.status(cliCmd, user)
86	88		if err != nil {
	89	+	logger.Error("status cmd", "err", err)
87	90		sesh.Fatal(err)
88	91		}
89	92		return next(sesh)
90	93		case "rss":
91	94		err := handler.rss(cliCmd, user)
92	95		if err != nil {
	96	+	logger.Error("rss cmd", "err", err)
93	97		sesh.Fatal(err)
94	98		}
95	99		return next(sesh)

...	...		@@ -341,6 +348,7 @@ func (handler CliHandler) monitor(cmd CliCmd, user *db.User) error {
341	348		resolvedTopic := result.Name
342	349
343	350		if *del {
	351	+	handler.Logger.Info("removing pipe monitor", "topic", resolvedTopic)
344	352		err := handler.DBPool.RemovePipeMonitor(user.ID, resolvedTopic)
345	353		if err != nil {
346	354		return fmt.Errorf("failed to delete monitor: %w", err)

pico

Patchset 103.2 on 2026-01-08T01:19:34Z · commit 74fb40f

Patchset `103.2` on 2026-01-08T01:19:34Z · commit 74fb40f