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 10ffdf1

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

feat(pipe): uptime cli command

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

Semantic diff summary

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

pkg/apps/pipe/cli.go

function_declaration Middleware modified
method_declaration uptime added
function_declaration parseDuration added

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

......@@ -90,6 +90,13 @@ func Middleware(handler *CliHandler) pssh.SSHServerMiddleware {
					sesh.Fatal(err)
				}
				return next(sesh)
			case "uptime":
				err := handler.uptime(cliCmd, user)
				if err != nil {
					logger.Error("uptime cmd", "err", err)
					sesh.Fatal(err)
				}
				return next(sesh)
			case "rss":
				err := handler.rss(cliCmd, user)
				if err != nil {

......@@ -442,6 +449,92 @@ func (handler *CliHandler) status(cmd *CliCmd, user *db.User) error {
	return nil
}

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

	fs := flag.NewFlagSet("uptime", flag.ContinueOnError)
	fs.SetOutput(cmd.sesh)
	fromStr := fs.String("from", "", "start time (RFC3339 or duration like '24h', '7d')")
	toStr := fs.String("to", "", "end time (RFC3339, defaults to now)")

	if err := fs.Parse(cmd.args); err != nil {
		return nil
	}

	args := fs.Args()
	if len(args) == 0 {
		_, _ = fmt.Fprintln(cmd.sesh, "usage: uptime <topic> [--from <time>] [--to <time>]")
		_, _ = fmt.Fprintln(cmd.sesh, "  --from: start time (RFC3339 or duration like '24h', '7d', default: 24h)")
		_, _ = fmt.Fprintln(cmd.sesh, "  --to: end time (RFC3339, default: now)")
		return nil
	}

	topic := args[0]

	monitor, err := handler.DBPool.FindPipeMonitorByTopic(user.ID, topic)
	if err != nil {
		return fmt.Errorf("failed to find monitor: %w", err)
	}
	if monitor == nil {
		return fmt.Errorf("monitor not found: %s", topic)
	}

	now := time.Now().UTC()
	to := now
	from := now.Add(-24 * time.Hour)

	if *fromStr != "" {
		if parsed, err := time.Parse(time.RFC3339, *fromStr); err == nil {
			from = parsed.UTC()
		} else if dur, err := parseDuration(*fromStr); err == nil {
			from = now.Add(-dur)
		} else {
			return fmt.Errorf("invalid --from value: %s", *fromStr)
		}
	}

	if *toStr != "" {
		if parsed, err := time.Parse(time.RFC3339, *toStr); err == nil {
			to = parsed.UTC()
		} else {
			return fmt.Errorf("invalid --to value: %s", *toStr)
		}
	}

	history, err := handler.DBPool.FindPipeMonitorHistory(monitor.ID, from, to)
	if err != nil {
		return fmt.Errorf("failed to fetch history: %w", err)
	}

	result := db.ComputeUptime(history, from, to)

	_, _ = fmt.Fprintf(cmd.sesh, "Monitor: %s\r\n", topic)
	_, _ = fmt.Fprintf(cmd.sesh, "Period: %s to %s\r\n", from.Format(time.RFC3339), to.Format(time.RFC3339))
	_, _ = fmt.Fprintf(cmd.sesh, "Total Duration: %s\r\n", result.TotalDuration.Round(time.Second))
	_, _ = fmt.Fprintf(cmd.sesh, "Uptime Duration: %s\r\n", result.UptimeDuration.Round(time.Second))
	_, _ = fmt.Fprintf(cmd.sesh, "Uptime: %.2f%%\r\n", result.UptimePercent)

	return nil
}

func parseDuration(s string) (time.Duration, error) {
	if len(s) == 0 {
		return 0, fmt.Errorf("empty duration")
	}
	last := s[len(s)-1]
	if last == 'd' {
		var n int
		_, err := fmt.Sscanf(s, "%d", &n)
		if err != nil {
			return 0, fmt.Errorf("invalid duration: %s", s)
		}
		return time.Duration(n) * 24 * time.Hour, nil
	}
	return time.ParseDuration(s)
}

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

...	...		@@ -442,6 +449,92 @@ func (handler CliHandler) status(cmd CliCmd, user *db.User) error {
442	449		return nil
443	450		}
444	451
	452	+	func (handler CliHandler) uptime(cmd CliCmd, user *db.User) error {
	453	+	if user == nil {
	454	+	return fmt.Errorf("access denied")
	455	+	}
	456	+
	457	+	fs := flag.NewFlagSet("uptime", flag.ContinueOnError)
	458	+	fs.SetOutput(cmd.sesh)
	459	+	fromStr := fs.String("from", "", "start time (RFC3339 or duration like '24h', '7d')")
	460	+	toStr := fs.String("to", "", "end time (RFC3339, defaults to now)")
	461	+
	462	+	if err := fs.Parse(cmd.args); err != nil {
	463	+	return nil
	464	+	}
	465	+
	466	+	args := fs.Args()
	467	+	if len(args) == 0 {
	468	+	_, _ = fmt.Fprintln(cmd.sesh, "usage: uptime <topic> [--from <time>] [--to <time>]")
	469	+	_, _ = fmt.Fprintln(cmd.sesh, " --from: start time (RFC3339 or duration like '24h', '7d', default: 24h)")
	470	+	_, _ = fmt.Fprintln(cmd.sesh, " --to: end time (RFC3339, default: now)")
	471	+	return nil
	472	+	}
	473	+
	474	+	topic := args[0]
	475	+
	476	+	monitor, err := handler.DBPool.FindPipeMonitorByTopic(user.ID, topic)
	477	+	if err != nil {
	478	+	return fmt.Errorf("failed to find monitor: %w", err)
	479	+	}
	480	+	if monitor == nil {
	481	+	return fmt.Errorf("monitor not found: %s", topic)
	482	+	}
	483	+
	484	+	now := time.Now().UTC()
	485	+	to := now
	486	+	from := now.Add(-24 * time.Hour)
	487	+
	488	+	if *fromStr != "" {
	489	+	if parsed, err := time.Parse(time.RFC3339, *fromStr); err == nil {
	490	+	from = parsed.UTC()
	491	+	} else if dur, err := parseDuration(*fromStr); err == nil {
	492	+	from = now.Add(-dur)
	493	+	} else {
	494	+	return fmt.Errorf("invalid --from value: %s", *fromStr)
	495	+	}
	496	+	}
	497	+
	498	+	if *toStr != "" {
	499	+	if parsed, err := time.Parse(time.RFC3339, *toStr); err == nil {
	500	+	to = parsed.UTC()
	501	+	} else {
	502	+	return fmt.Errorf("invalid --to value: %s", *toStr)
	503	+	}
	504	+	}
	505	+
	506	+	history, err := handler.DBPool.FindPipeMonitorHistory(monitor.ID, from, to)
	507	+	if err != nil {
	508	+	return fmt.Errorf("failed to fetch history: %w", err)
	509	+	}
	510	+
	511	+	result := db.ComputeUptime(history, from, to)
	512	+
	513	+	_, _ = fmt.Fprintf(cmd.sesh, "Monitor: %s\r\n", topic)
	514	+	_, _ = fmt.Fprintf(cmd.sesh, "Period: %s to %s\r\n", from.Format(time.RFC3339), to.Format(time.RFC3339))
	515	+	_, _ = fmt.Fprintf(cmd.sesh, "Total Duration: %s\r\n", result.TotalDuration.Round(time.Second))
	516	+	_, _ = fmt.Fprintf(cmd.sesh, "Uptime Duration: %s\r\n", result.UptimeDuration.Round(time.Second))
	517	+	_, _ = fmt.Fprintf(cmd.sesh, "Uptime: %.2f%%\r\n", result.UptimePercent)
	518	+
	519	+	return nil
	520	+	}
	521	+
	522	+	func parseDuration(s string) (time.Duration, error) {
	523	+	if len(s) == 0 {
	524	+	return 0, fmt.Errorf("empty duration")
	525	+	}
	526	+	last := s[len(s)-1]
	527	+	if last == 'd' {
	528	+	var n int
	529	+	_, err := fmt.Sscanf(s, "%d", &n)
	530	+	if err != nil {
	531	+	return 0, fmt.Errorf("invalid duration: %s", s)
	532	+	}
	533	+	return time.Duration(n) * 24 * time.Hour, nil
	534	+	}
	535	+	return time.ParseDuration(s)
	536	+	}
	537	+
445	538		func (handler CliHandler) rss(cmd CliCmd, user *db.User) error {
446	539		if user == nil {
447	540		return fmt.Errorf("access denied")

...	...		@@ -90,6 +90,13 @@ func Middleware(handler *CliHandler) pssh.SSHServerMiddleware {
90	90		sesh.Fatal(err)
91	91		}
92	92		return next(sesh)
	93	+	case "uptime":
	94	+	err := handler.uptime(cliCmd, user)
	95	+	if err != nil {
	96	+	logger.Error("uptime cmd", "err", err)
	97	+	sesh.Fatal(err)
	98	+	}
	99	+	return next(sesh)
93	100		case "rss":
94	101		err := handler.rss(cliCmd, user)
95	102		if err != nil {

pico

Patchset 103.2 on 2026-01-08T01:19:34Z · commit 10ffdf1

Patchset `103.2` on 2026-01-08T01:19:34Z · commit 10ffdf1