pico

feat(pipe): add pipe_monitors table rss
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 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
+69 -0 pkg/apps/pipe/cli.go #
......@@ -71,6 +71,12 @@ func Middleware(handler *CliHandler) pssh.SSHServerMiddleware {
7171 sesh.Fatal(err)
7272 }
7373 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)
7480 }
7581
7682 topic := ""
......@@ -274,6 +280,69 @@ func (handler *CliHandler) ls(cmd *CliCmd) error {
274280 return nil
275281 }
276282
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+
277346 func (handler *CliHandler) pub(cmd *CliCmd, topic string, clientID string) error {
278347 pubCmd := flagSet("pub", cmd.sesh)
279348 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 {
503503 UpdatePipeMonitorLastPing(userID, topic string, lastPing *time.Time) error
504504 RemovePipeMonitor(userID, topic string) error
505505 FindPipeMonitorByTopic(userID, topic string) (*PipeMonitor, error)
506+ FindPipeMonitorsByUser(userID string) ([]*PipeMonitor, error)
506507
507508 Close() error
508509 }
+9 -0 pkg/db/postgres/storage.go #
......@@ -1556,3 +1556,12 @@ func (me *PsqlDB) FindPipeMonitorByTopic(userID, topic string) (*db.PipeMonitor,
15561556 }
15571557 return monitor, nil
15581558 }
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+}
+46 -0 pkg/db/postgres/storage_test.go #
......@@ -1582,3 +1582,49 @@ func TestFindPipeMonitorByTopic_NotFound(t *testing.T) {
15821582 t.Error("expected error for nonexistent monitor, got nil")
15831583 }
15841584 }
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+}
+4 -0 pkg/db/stub/stub.go #
......@@ -259,3 +259,7 @@ func (me *StubDB) RemovePipeMonitor(userID, topic string) error {
259259 func (me *StubDB) FindPipeMonitorByTopic(userID, topic string) (*db.PipeMonitor, error) {
260260 return nil, errNotImpl
261261 }
262+
263+func (me *StubDB) FindPipeMonitorsByUser(userID string) ([]*db.PipeMonitor, error) {
264+ return nil, errNotImpl
265+}
Back to top