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 8df2aa2

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
+43 -4 pkg/apps/pipe/cli.go #
......@@ -60,6 +60,7 @@ func Middleware(handler *CliHandler) pssh.SSHServerMiddleware {
6060 isAdmin: isAdmin,
6161 pipeCtx: pipeCtx,
6262 cancel: cancel,
63+ user: user,
6364 }
6465
6566 cmd := strings.TrimSpace(args[0])
......@@ -181,6 +182,7 @@ type CliCmd struct {
181182 isAdmin bool
182183 pipeCtx context.Context
183184 cancel context.CancelFunc
185+ user *db.User
184186 }
185187
186188 func help(cfg *shared.ConfigSite, sesh *pssh.SSHServerConnSession) {
......@@ -320,12 +322,21 @@ func (handler *CliHandler) monitor(cmd *CliCmd, user *db.User) error {
320322 return fmt.Errorf("topic is required")
321323 }
322324
325+ // Resolve to fully qualified topic name
326+ result := resolveTopic(TopicResolveInput{
327+ UserName: cmd.userName,
328+ Topic: topic,
329+ IsAdmin: cmd.isAdmin,
330+ IsPublic: false,
331+ })
332+ resolvedTopic := result.Name
333+
323334 if *del {
324- err := handler.DBPool.RemovePipeMonitor(user.ID, topic)
335+ err := handler.DBPool.RemovePipeMonitor(user.ID, resolvedTopic)
325336 if err != nil {
326337 return fmt.Errorf("failed to delete monitor: %w", err)
327338 }
328- _, _ = fmt.Fprintf(cmd.sesh, "monitor deleted: %s\r\n", topic)
339+ _, _ = fmt.Fprintf(cmd.sesh, "monitor deleted: %s\r\n", resolvedTopic)
329340 return nil
330341 }
331342
......@@ -348,12 +359,12 @@ func (handler *CliHandler) monitor(cmd *CliCmd, user *db.User) error {
348359 }
349360
350361 winEnd := time.Now().Add(dur)
351- err = handler.DBPool.UpsertPipeMonitor(user.ID, topic, dur, &winEnd)
362+ err = handler.DBPool.UpsertPipeMonitor(user.ID, resolvedTopic, dur, &winEnd)
352363 if err != nil {
353364 return fmt.Errorf("failed to create monitor: %w", err)
354365 }
355366
356- _, _ = fmt.Fprintf(cmd.sesh, "monitor created: %s (window: %s)\r\n", topic, dur)
367+ _, _ = fmt.Fprintf(cmd.sesh, "monitor created: %s (window: %s)\r\n", resolvedTopic, dur)
357368 return nil
358369 }
359370
......@@ -672,9 +683,35 @@ func (handler *CliHandler) pub(cmd *CliCmd, topic string, clientID string) error
672683 return err
673684 }
674685
686+ handler.updateMonitor(cmd, name)
687+
675688 return nil
676689 }
677690
691+func (handler *CliHandler) updateMonitor(cmd *CliCmd, topic string) {
692+ if cmd.user == nil {
693+ return
694+ }
695+
696+ monitor, err := handler.DBPool.FindPipeMonitorByTopic(cmd.user.ID, topic)
697+ if err != nil || monitor == nil {
698+ return
699+ }
700+
701+ now := time.Now()
702+ if err := handler.DBPool.UpdatePipeMonitorLastPing(cmd.user.ID, topic, &now); err != nil {
703+ handler.Logger.Error("failed to update monitor last_ping", "err", err, "topic", topic)
704+ }
705+
706+ // Advance window if current time is past window_end
707+ if monitor.WindowEnd != nil && now.After(*monitor.WindowEnd) {
708+ nextWindow := monitor.GetNextWindow()
709+ if err := handler.DBPool.UpsertPipeMonitor(cmd.user.ID, topic, monitor.WindowDur, nextWindow); err != nil {
710+ handler.Logger.Error("failed to advance monitor window", "err", err, "topic", topic)
711+ }
712+ }
713+}
714+
678715 func (handler *CliHandler) sub(cmd *CliCmd, topic string, clientID string) error {
679716 subCmd := flagSet("sub", cmd.sesh)
680717 access := subCmd.String("a", "", "Comma separated list of pico usernames or ssh-key fingerprints to allow access to a topic")
......@@ -872,6 +909,8 @@ func (handler *CliHandler) pipe(cmd *CliCmd, topic string, clientID string) erro
872909 return writeErr
873910 }
874911
912+ handler.updateMonitor(cmd, name)
913+
875914 return nil
876915 }
877916
+13 -13 pkg/apps/pipe/ssh_test.go #
......@@ -1505,8 +1505,8 @@ func TestMonitor_CreateMonitor(t *testing.T) {
15051505 t.Errorf("authenticated user should not get access denied, got: %s", output)
15061506 }
15071507
1508- // Verify monitor was created in DB
1509- monitor, err := server.DBPool.FindPipeMonitorByTopic("alice-id", "pico-uptime")
1508+ // Verify monitor was created in DB (topic is stored with user prefix)
1509+ monitor, err := server.DBPool.FindPipeMonitorByTopic("alice-id", "alice/pico-uptime")
15101510 if err != nil {
15111511 t.Fatalf("monitor should exist in DB: %v", err)
15121512 }
......@@ -1515,7 +1515,7 @@ func TestMonitor_CreateMonitor(t *testing.T) {
15151515 t.Errorf("expected window duration 24h, got: %v", monitor.WindowDur)
15161516 }
15171517
1518- if !strings.Contains(output, "pico-uptime") || !strings.Contains(output, "24h") {
1518+ if !strings.Contains(output, "alice/pico-uptime") || !strings.Contains(output, "24h") {
15191519 t.Errorf("output should confirm monitor creation, got: %s", output)
15201520 }
15211521 }
......@@ -1545,8 +1545,8 @@ func TestMonitor_UpdateMonitor(t *testing.T) {
15451545 t.Logf("update command completed: %v", err)
15461546 }
15471547
1548- // Verify monitor was updated
1549- monitor, err := server.DBPool.FindPipeMonitorByTopic("alice-id", "my-cron")
1548+ // Verify monitor was updated (topic is stored with user prefix)
1549+ monitor, err := server.DBPool.FindPipeMonitorByTopic("alice-id", "alice/my-cron")
15501550 if err != nil {
15511551 t.Fatalf("monitor should exist in DB: %v", err)
15521552 }
......@@ -1579,8 +1579,8 @@ func TestMonitor_DeleteMonitor(t *testing.T) {
15791579 t.Logf("create command completed: %v", err)
15801580 }
15811581
1582- // Verify it exists
1583- _, err = server.DBPool.FindPipeMonitorByTopic("alice-id", "to-delete")
1582+ // Verify it exists (topic is stored with user prefix)
1583+ _, err = server.DBPool.FindPipeMonitorByTopic("alice-id", "alice/to-delete")
15841584 if err != nil {
15851585 t.Fatalf("monitor should exist before deletion: %v", err)
15861586 }
......@@ -1591,8 +1591,8 @@ func TestMonitor_DeleteMonitor(t *testing.T) {
15911591 t.Logf("delete command completed: %v", err)
15921592 }
15931593
1594- // Verify it's gone
1595- _, err = server.DBPool.FindPipeMonitorByTopic("alice-id", "to-delete")
1594+ // Verify it's gone (topic is stored with user prefix)
1595+ _, err = server.DBPool.FindPipeMonitorByTopic("alice-id", "alice/to-delete")
15961596 if err == nil {
15971597 t.Errorf("monitor should be deleted from DB")
15981598 }
......@@ -1873,10 +1873,10 @@ func TestPub_UpdatesMonitorLastPing(t *testing.T) {
18731873 user := GenerateUser("alice")
18741874 RegisterUserWithServer(server, user)
18751875
1876- // Create a monitor first
1876+ // Create a monitor first (topic is stored with user prefix)
18771877 now := time.Now()
18781878 windowEnd := now.Add(1 * time.Hour)
1879- _ = server.DBPool.UpsertPipeMonitor("alice-id", "ping-test", 1*time.Hour, &windowEnd)
1879+ _ = server.DBPool.UpsertPipeMonitor("alice-id", "alice/ping-test", 1*time.Hour, &windowEnd)
18801880
18811881 subClient, err := user.NewClient()
18821882 if err != nil {
......@@ -1909,8 +1909,8 @@ func TestPub_UpdatesMonitorLastPing(t *testing.T) {
19091909 t.Logf("pub command completed: %v", err)
19101910 }
19111911
1912- // Verify last_ping was updated
1913- monitor, err := server.DBPool.FindPipeMonitorByTopic("alice-id", "ping-test")
1912+ // Verify last_ping was updated (topic is stored with user prefix)
1913+ monitor, err := server.DBPool.FindPipeMonitorByTopic("alice-id", "alice/ping-test")
19141914 if err != nil {
19151915 t.Fatalf("monitor should exist: %v", err)
19161916 }
Back to top