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 43aa4a1

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
+42 -0 pkg/db/postgres/storage.go #
......@@ -1514,3 +1514,45 @@ func (me *PsqlDB) InsertAccessLog(log *db.AccessLog) error {
15141514 )
15151515 return err
15161516 }
1517+
1518+func (me *PsqlDB) UpsertPipeMonitor(userID, topic string, dur time.Duration, winEnd *time.Time) error {
1519+ durStr := fmt.Sprintf("%d seconds", int64(dur.Seconds()))
1520+ _, err := me.Db.Exec(
1521+ `INSERT INTO pipe_monitors (user_id, topic, window_dur, window_end)
1522+ VALUES ($1, $2, $3::interval, $4)
1523+ ON CONFLICT (user_id, topic) DO UPDATE SET window_dur = $3::interval, window_end = $4, updated_at = NOW();`,
1524+ userID,
1525+ topic,
1526+ durStr,
1527+ winEnd,
1528+ )
1529+ return err
1530+}
1531+
1532+func (me *PsqlDB) UpdatePipeMonitorLastPing(userID, topic string, lastPing *time.Time) error {
1533+ _, err := me.Db.Exec(
1534+ `UPDATE pipe_monitors SET last_ping = $3, updated_at = NOW() WHERE user_id = $1 AND topic = $2;`,
1535+ userID,
1536+ topic,
1537+ lastPing,
1538+ )
1539+ return err
1540+}
1541+
1542+func (me *PsqlDB) RemovePipeMonitor(userID, topic string) error {
1543+ _, err := me.Db.Exec(
1544+ `DELETE FROM pipe_monitors WHERE user_id = $1 AND topic = $2;`,
1545+ userID,
1546+ topic,
1547+ )
1548+ return err
1549+}
1550+
1551+func (me *PsqlDB) FindPipeMonitorByTopic(userID, topic string) (*db.PipeMonitor, error) {
1552+ monitor := &db.PipeMonitor{}
1553+ err := me.Db.Get(monitor, `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 AND topic = $2;`, userID, topic)
1554+ if err != nil {
1555+ return nil, err
1556+ }
1557+ return monitor, nil
1558+}
+111 -1 pkg/db/postgres/storage_test.go #
......@@ -169,7 +169,7 @@ func cleanupTestData(t *testing.T) {
169169 "access_logs", "tuns_event_logs", "analytics_visits",
170170 "feed_items", "post_aliases", "post_tags", "posts",
171171 "projects", "feature_flags", "payment_history", "tokens",
172- "public_keys", "app_users",
172+ "public_keys", "pipe_monitors", "app_users",
173173 }
174174 for _, table := range tables {
175175 _, err := testDB.Db.Exec(fmt.Sprintf("DELETE FROM %s", table))
......@@ -1472,3 +1472,113 @@ func TestPaymentHistoryData_JSONBRoundtrip(t *testing.T) {
14721472 t.Errorf("expected tx_id 'tx789', got '%s'", txId)
14731473 }
14741474 }
1475+
1476+// ============ Pipe Monitor Tests ============
1477+
1478+func TestUpsertPipeMonitor(t *testing.T) {
1479+ cleanupTestData(t)
1480+
1481+ user, _ := testDB.RegisterUser("pipemonitorowner", "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAI pipemonitorowner", "comment")
1482+
1483+ winEnd := time.Now().Add(time.Hour)
1484+ err := testDB.UpsertPipeMonitor(user.ID, "test-topic", 5*time.Minute, &winEnd)
1485+ if err != nil {
1486+ t.Fatalf("UpsertPipeMonitor failed: %v", err)
1487+ }
1488+
1489+ monitor, err := testDB.FindPipeMonitorByTopic(user.ID, "test-topic")
1490+ if err != nil {
1491+ t.Fatalf("FindPipeMonitorByTopic failed: %v", err)
1492+ }
1493+ if monitor.Topic != "test-topic" {
1494+ t.Errorf("expected topic 'test-topic', got '%s'", monitor.Topic)
1495+ }
1496+ if monitor.WindowDur != 5*time.Minute {
1497+ t.Errorf("expected window_dur 5m, got %v", monitor.WindowDur)
1498+ }
1499+}
1500+
1501+func TestUpsertPipeMonitor_Update(t *testing.T) {
1502+ cleanupTestData(t)
1503+
1504+ user, _ := testDB.RegisterUser("pipeupdateowner", "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAI pipeupdateowner", "comment")
1505+
1506+ winEnd1 := time.Now().Add(time.Hour)
1507+ err := testDB.UpsertPipeMonitor(user.ID, "update-topic", 5*time.Minute, &winEnd1)
1508+ if err != nil {
1509+ t.Fatalf("first UpsertPipeMonitor failed: %v", err)
1510+ }
1511+
1512+ winEnd2 := time.Now().Add(2 * time.Hour)
1513+ err = testDB.UpsertPipeMonitor(user.ID, "update-topic", 10*time.Minute, &winEnd2)
1514+ if err != nil {
1515+ t.Fatalf("second UpsertPipeMonitor failed: %v", err)
1516+ }
1517+
1518+ monitor, err := testDB.FindPipeMonitorByTopic(user.ID, "update-topic")
1519+ if err != nil {
1520+ t.Fatalf("FindPipeMonitorByTopic failed: %v", err)
1521+ }
1522+ if monitor.WindowDur != 10*time.Minute {
1523+ t.Errorf("expected window_dur 10m after update, got %v", monitor.WindowDur)
1524+ }
1525+}
1526+
1527+func TestUpdatePipeMonitorLastPing(t *testing.T) {
1528+ cleanupTestData(t)
1529+
1530+ user, _ := testDB.RegisterUser("pipepingowner", "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAI pipepingowner", "comment")
1531+
1532+ winEnd := time.Now().Add(time.Hour)
1533+ err := testDB.UpsertPipeMonitor(user.ID, "ping-topic", 5*time.Minute, &winEnd)
1534+ if err != nil {
1535+ t.Fatalf("UpsertPipeMonitor failed: %v", err)
1536+ }
1537+
1538+ lastPing := time.Now()
1539+ err = testDB.UpdatePipeMonitorLastPing(user.ID, "ping-topic", &lastPing)
1540+ if err != nil {
1541+ t.Fatalf("UpdatePipeMonitorLastPing failed: %v", err)
1542+ }
1543+
1544+ monitor, err := testDB.FindPipeMonitorByTopic(user.ID, "ping-topic")
1545+ if err != nil {
1546+ t.Fatalf("FindPipeMonitorByTopic failed: %v", err)
1547+ }
1548+ if monitor.LastPing == nil {
1549+ t.Error("expected last_ping to be set, got nil")
1550+ }
1551+}
1552+
1553+func TestRemovePipeMonitor(t *testing.T) {
1554+ cleanupTestData(t)
1555+
1556+ user, _ := testDB.RegisterUser("piperemoveowner", "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAI piperemoveowner", "comment")
1557+
1558+ winEnd := time.Now().Add(time.Hour)
1559+ err := testDB.UpsertPipeMonitor(user.ID, "remove-topic", 5*time.Minute, &winEnd)
1560+ if err != nil {
1561+ t.Fatalf("UpsertPipeMonitor failed: %v", err)
1562+ }
1563+
1564+ err = testDB.RemovePipeMonitor(user.ID, "remove-topic")
1565+ if err != nil {
1566+ t.Fatalf("RemovePipeMonitor failed: %v", err)
1567+ }
1568+
1569+ _, err = testDB.FindPipeMonitorByTopic(user.ID, "remove-topic")
1570+ if err == nil {
1571+ t.Error("expected error finding removed monitor, got nil")
1572+ }
1573+}
1574+
1575+func TestFindPipeMonitorByTopic_NotFound(t *testing.T) {
1576+ cleanupTestData(t)
1577+
1578+ user, _ := testDB.RegisterUser("pipenotfoundowner", "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAI pipenotfoundowner", "comment")
1579+
1580+ _, err := testDB.FindPipeMonitorByTopic(user.ID, "nonexistent-topic")
1581+ if err == nil {
1582+ t.Error("expected error for nonexistent monitor, got nil")
1583+ }
1584+}
+16 -0 pkg/db/stub/stub.go #
......@@ -243,3 +243,19 @@ func (me *StubDB) FindPubkeysInAccessLogs(userID string) ([]string, error) {
243243 func (me *StubDB) InsertAccessLog(log *db.AccessLog) error {
244244 return errNotImpl
245245 }
246+
247+func (me *StubDB) UpsertPipeMonitor(userID, topic string, dur time.Duration, winEnd *time.Time) error {
248+ return errNotImpl
249+}
250+
251+func (me *StubDB) UpdatePipeMonitorLastPing(userID, topic string, lastPing *time.Time) error {
252+ return errNotImpl
253+}
254+
255+func (me *StubDB) RemovePipeMonitor(userID, topic string) error {
256+ return errNotImpl
257+}
258+
259+func (me *StubDB) FindPipeMonitorByTopic(userID, topic string) (*db.PipeMonitor, error) {
260+ return nil, errNotImpl
261+}
Back to top