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 9c7fcda

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): add monitor cli

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

Semantic diff summary

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

pkg/apps/pipe/cli.go

function_declaration Middleware modified
method_declaration monitor added

pkg/db/db.go

chunk lines 503-509 modified

pkg/db/postgres/storage.go

method_declaration FindPipeMonitorsByUser added

pkg/db/postgres/storage_test.go

function_declaration TestFindPipeMonitorsByUser added
function_declaration TestFindPipeMonitorsByUser_Empty added

pkg/db/stub/stub.go

method_declaration FindPipeMonitorsByUser added

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

......@@ -71,6 +71,12 @@ func Middleware(handler *CliHandler) pssh.SSHServerMiddleware {
					sesh.Fatal(err)
				}
				return next(sesh)
			case "monitor":
				err := handler.monitor(cliCmd, user)
				if err != nil {
					sesh.Fatal(err)
				}
				return next(sesh)
			}

			topic := ""

......@@ -274,6 +280,69 @@ func (handler *CliHandler) ls(cmd *CliCmd) error {
	return nil
}

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

	args := cmd.sesh.Command()
	topic := ""
	cmdArgs := args[1:]
	if len(args) > 1 && !strings.HasPrefix(args[1], "-") {
		topic = strings.TrimSpace(args[1])
		cmdArgs = args[2:]
	}

	monitorCmd := flagSet("monitor", cmd.sesh)
	del := monitorCmd.Bool("d", false, "Delete the monitor")

	if !flagCheck(monitorCmd, topic, cmdArgs) {
		return nil
	}

	if topic == "" {
		_, _ = fmt.Fprintln(cmd.sesh, "Usage: monitor <topic> <duration>")
		_, _ = fmt.Fprintln(cmd.sesh, "       monitor <topic> -d")
		return fmt.Errorf("topic is required")
	}

	if *del {
		err := handler.DBPool.RemovePipeMonitor(user.ID, topic)
		if err != nil {
			return fmt.Errorf("failed to delete monitor: %w", err)
		}
		_, _ = fmt.Fprintf(cmd.sesh, "monitor deleted: %s\r\n", topic)
		return nil
	}

	// Create/update monitor - need duration argument
	durStr := ""
	if monitorCmd.NArg() > 0 {
		durStr = monitorCmd.Arg(0)
	} else if len(cmdArgs) > 0 {
		durStr = cmdArgs[0]
	}

	if durStr == "" {
		_, _ = fmt.Fprintln(cmd.sesh, "Usage: monitor <topic> <duration>")
		return fmt.Errorf("duration is required")
	}

	dur, err := time.ParseDuration(durStr)
	if err != nil {
		return fmt.Errorf("invalid duration %q: %w", durStr, err)
	}

	winEnd := time.Now().Add(dur)
	err = handler.DBPool.UpsertPipeMonitor(user.ID, topic, dur, &winEnd)
	if err != nil {
		return fmt.Errorf("failed to create monitor: %w", err)
	}

	_, _ = fmt.Fprintf(cmd.sesh, "monitor created: %s (window: %s)\r\n", topic, dur)
	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")

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

......@@ -503,6 +503,7 @@ type DB interface {
	UpdatePipeMonitorLastPing(userID, topic string, lastPing *time.Time) error
	RemovePipeMonitor(userID, topic string) error
	FindPipeMonitorByTopic(userID, topic string) (*PipeMonitor, error)
	FindPipeMonitorsByUser(userID string) ([]*PipeMonitor, error)

	Close() error
}

+9 -0 pkg/db/postgres/storage.go #

......@@ -1556,3 +1556,12 @@ func (me *PsqlDB) FindPipeMonitorByTopic(userID, topic string) (*db.PipeMonitor,
	}
	return monitor, nil
}

func (me *PsqlDB) FindPipeMonitorsByUser(userID string) ([]*db.PipeMonitor, error) {
	var monitors []*db.PipeMonitor
	err := me.Db.Select(&monitors, `SELECT id, user_id, topic, (EXTRACT(EPOCH FROM window_dur) * 1000000000)::bigint as window_dur, window_end, last_ping, created_at, updated_at FROM pipe_monitors WHERE user_id = $1 ORDER BY topic;`, userID)
	if err != nil {
		return nil, err
	}
	return monitors, nil
}

+46 -0 pkg/db/postgres/storage_test.go #

......@@ -1582,3 +1582,49 @@ func TestFindPipeMonitorByTopic_NotFound(t *testing.T) {
		t.Error("expected error for nonexistent monitor, got nil")
	}
}

func TestFindPipeMonitorsByUser(t *testing.T) {
	cleanupTestData(t)

	user, _ := testDB.RegisterUser("pipemonlistowner", "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAI pipemonlistowner", "comment")

	winEnd := time.Now().Add(time.Hour)
	_ = testDB.UpsertPipeMonitor(user.ID, "service-a", 5*time.Minute, &winEnd)
	_ = testDB.UpsertPipeMonitor(user.ID, "service-b", 10*time.Minute, &winEnd)
	_ = testDB.UpsertPipeMonitor(user.ID, "service-c", 1*time.Hour, &winEnd)

	monitors, err := testDB.FindPipeMonitorsByUser(user.ID)
	if err != nil {
		t.Fatalf("FindPipeMonitorsByUser failed: %v", err)
	}

	if len(monitors) != 3 {
		t.Errorf("expected 3 monitors, got %d", len(monitors))
	}

	// Should be ordered by topic
	if monitors[0].Topic != "service-a" {
		t.Errorf("expected first topic 'service-a', got %s", monitors[0].Topic)
	}
	if monitors[1].Topic != "service-b" {
		t.Errorf("expected second topic 'service-b', got %s", monitors[1].Topic)
	}
	if monitors[2].Topic != "service-c" {
		t.Errorf("expected third topic 'service-c', got %s", monitors[2].Topic)
	}
}

func TestFindPipeMonitorsByUser_Empty(t *testing.T) {
	cleanupTestData(t)

	user, _ := testDB.RegisterUser("pipenomonitors", "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAI pipenomonitors", "comment")

	monitors, err := testDB.FindPipeMonitorsByUser(user.ID)
	if err != nil {
		t.Fatalf("FindPipeMonitorsByUser failed: %v", err)
	}

	if len(monitors) != 0 {
		t.Errorf("expected 0 monitors for user with none, got %d", len(monitors))
	}
}

+4 -0 pkg/db/stub/stub.go #

......@@ -259,3 +259,7 @@ func (me *StubDB) RemovePipeMonitor(userID, topic string) error {
func (me *StubDB) FindPipeMonitorByTopic(userID, topic string) (*db.PipeMonitor, error) {
	return nil, errNotImpl
}

func (me *StubDB) FindPipeMonitorsByUser(userID string) ([]*db.PipeMonitor, error) {
	return nil, errNotImpl
}

...	...		@@ -274,6 +280,69 @@ func (handler CliHandler) ls(cmd CliCmd) error {
274	280		return nil
275	281		}
276	282
	283	+	func (handler CliHandler) monitor(cmd CliCmd, user *db.User) error {
	284	+	if user == nil {
	285	+	return fmt.Errorf("access denied")
	286	+	}
	287	+
	288	+	args := cmd.sesh.Command()
	289	+	topic := ""
	290	+	cmdArgs := args[1:]
	291	+	if len(args) > 1 && !strings.HasPrefix(args[1], "-") {
	292	+	topic = strings.TrimSpace(args[1])
	293	+	cmdArgs = args[2:]
	294	+	}
	295	+
	296	+	monitorCmd := flagSet("monitor", cmd.sesh)
	297	+	del := monitorCmd.Bool("d", false, "Delete the monitor")
	298	+
	299	+	if !flagCheck(monitorCmd, topic, cmdArgs) {
	300	+	return nil
	301	+	}
	302	+
	303	+	if topic == "" {
	304	+	_, _ = fmt.Fprintln(cmd.sesh, "Usage: monitor <topic> <duration>")
	305	+	_, _ = fmt.Fprintln(cmd.sesh, " monitor <topic> -d")
	306	+	return fmt.Errorf("topic is required")
	307	+	}
	308	+
	309	+	if *del {
	310	+	err := handler.DBPool.RemovePipeMonitor(user.ID, topic)
	311	+	if err != nil {
	312	+	return fmt.Errorf("failed to delete monitor: %w", err)
	313	+	}
	314	+	_, _ = fmt.Fprintf(cmd.sesh, "monitor deleted: %s\r\n", topic)
	315	+	return nil
	316	+	}
	317	+
	318	+	// Create/update monitor - need duration argument
	319	+	durStr := ""
	320	+	if monitorCmd.NArg() > 0 {
	321	+	durStr = monitorCmd.Arg(0)
	322	+	} else if len(cmdArgs) > 0 {
	323	+	durStr = cmdArgs[0]
	324	+	}
	325	+
	326	+	if durStr == "" {
	327	+	_, _ = fmt.Fprintln(cmd.sesh, "Usage: monitor <topic> <duration>")
	328	+	return fmt.Errorf("duration is required")
	329	+	}
	330	+
	331	+	dur, err := time.ParseDuration(durStr)
	332	+	if err != nil {
	333	+	return fmt.Errorf("invalid duration %q: %w", durStr, err)
	334	+	}
	335	+
	336	+	winEnd := time.Now().Add(dur)
	337	+	err = handler.DBPool.UpsertPipeMonitor(user.ID, topic, dur, &winEnd)
	338	+	if err != nil {
	339	+	return fmt.Errorf("failed to create monitor: %w", err)
	340	+	}
	341	+
	342	+	_, _ = fmt.Fprintf(cmd.sesh, "monitor created: %s (window: %s)\r\n", topic, dur)
	343	+	return nil
	344	+	}
	345	+
277	346		func (handler CliHandler) pub(cmd CliCmd, topic string, clientID string) error {
278	347		pubCmd := flagSet("pub", cmd.sesh)
279	348		access := pubCmd.String("a", "", "Comma separated list of pico usernames or ssh-key fingerprints to allow access to a topic")

...	...		@@ -1556,3 +1556,12 @@ func (me PsqlDB) FindPipeMonitorByTopic(userID, topic string) (db.PipeMonitor,
1556	1556		}
1557	1557		return monitor, nil
1558	1558		}
	1559	+
	1560	+	func (me PsqlDB) FindPipeMonitorsByUser(userID string) ([]db.PipeMonitor, error) {
	1561	+	var monitors []*db.PipeMonitor
	1562	+	err := me.Db.Select(&monitors, `SELECT id, user_id, topic, (EXTRACT(EPOCH FROM window_dur) * 1000000000)::bigint as window_dur, window_end, last_ping, created_at, updated_at FROM pipe_monitors WHERE user_id = $1 ORDER BY topic;`, userID)
	1563	+	if err != nil {
	1564	+	return nil, err
	1565	+	}
	1566	+	return monitors, nil
	1567	+	}

...	...		@@ -1582,3 +1582,49 @@ func TestFindPipeMonitorByTopic_NotFound(t *testing.T) {
1582	1582		t.Error("expected error for nonexistent monitor, got nil")
1583	1583		}
1584	1584		}
	1585	+
	1586	+	func TestFindPipeMonitorsByUser(t *testing.T) {
	1587	+	cleanupTestData(t)
	1588	+
	1589	+	user, _ := testDB.RegisterUser("pipemonlistowner", "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAI pipemonlistowner", "comment")
	1590	+
	1591	+	winEnd := time.Now().Add(time.Hour)
	1592	+	_ = testDB.UpsertPipeMonitor(user.ID, "service-a", 5*time.Minute, &winEnd)
	1593	+	_ = testDB.UpsertPipeMonitor(user.ID, "service-b", 10*time.Minute, &winEnd)
	1594	+	_ = testDB.UpsertPipeMonitor(user.ID, "service-c", 1*time.Hour, &winEnd)
	1595	+
	1596	+	monitors, err := testDB.FindPipeMonitorsByUser(user.ID)
	1597	+	if err != nil {
	1598	+	t.Fatalf("FindPipeMonitorsByUser failed: %v", err)
	1599	+	}
	1600	+
	1601	+	if len(monitors) != 3 {
	1602	+	t.Errorf("expected 3 monitors, got %d", len(monitors))
	1603	+	}
	1604	+
	1605	+	// Should be ordered by topic
	1606	+	if monitors[0].Topic != "service-a" {
	1607	+	t.Errorf("expected first topic 'service-a', got %s", monitors[0].Topic)
	1608	+	}
	1609	+	if monitors[1].Topic != "service-b" {
	1610	+	t.Errorf("expected second topic 'service-b', got %s", monitors[1].Topic)
	1611	+	}
	1612	+	if monitors[2].Topic != "service-c" {
	1613	+	t.Errorf("expected third topic 'service-c', got %s", monitors[2].Topic)
	1614	+	}
	1615	+	}
	1616	+
	1617	+	func TestFindPipeMonitorsByUser_Empty(t *testing.T) {
	1618	+	cleanupTestData(t)
	1619	+
	1620	+	user, _ := testDB.RegisterUser("pipenomonitors", "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAI pipenomonitors", "comment")
	1621	+
	1622	+	monitors, err := testDB.FindPipeMonitorsByUser(user.ID)
	1623	+	if err != nil {
	1624	+	t.Fatalf("FindPipeMonitorsByUser failed: %v", err)
	1625	+	}
	1626	+
	1627	+	if len(monitors) != 0 {
	1628	+	t.Errorf("expected 0 monitors for user with none, got %d", len(monitors))
	1629	+	}
	1630	+	}

...	...		@@ -71,6 +71,12 @@ func Middleware(handler *CliHandler) pssh.SSHServerMiddleware {
71	71		sesh.Fatal(err)
72	72		}
73	73		return next(sesh)
	74	+	case "monitor":
	75	+	err := handler.monitor(cliCmd, user)
	76	+	if err != nil {
	77	+	sesh.Fatal(err)
	78	+	}
	79	+	return next(sesh)
74	80		}
75	81
76	82		topic := ""

...	...		@@ -503,6 +503,7 @@ type DB interface {
503	503		UpdatePipeMonitorLastPing(userID, topic string, lastPing *time.Time) error
504	504		RemovePipeMonitor(userID, topic string) error
505	505		FindPipeMonitorByTopic(userID, topic string) (*PipeMonitor, error)
	506	+	FindPipeMonitorsByUser(userID string) ([]*PipeMonitor, error)
506	507
507	508		Close() error
508	509		}

...	...		@@ -259,3 +259,7 @@ func (me *StubDB) RemovePipeMonitor(userID, topic string) error {
259	259		func (me StubDB) FindPipeMonitorByTopic(userID, topic string) (db.PipeMonitor, error) {
260	260		return nil, errNotImpl
261	261		}
	262	+
	263	+	func (me StubDB) FindPipeMonitorsByUser(userID string) ([]db.PipeMonitor, error) {
	264	+	return nil, errNotImpl
	265	+	}

pico

Patchset 103.2 on 2026-01-08T01:19:34Z · commit 9c7fcda

Patchset `103.2` on 2026-01-08T01:19:34Z · commit 9c7fcda