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.2 on 2026-01-08T01:19:34Z · commit a75186b

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
+531 -3 pkg/apps/pipe/ssh_test.go #
......@@ -25,9 +25,10 @@ import (
2525
2626 type TestDB struct {
2727 *stub.StubDB
28- Users []*db.User
29- Pubkeys []*db.PublicKey
30- Features []*db.FeatureFlag
28+ Users []*db.User
29+ Pubkeys []*db.PublicKey
30+ Features []*db.FeatureFlag
31+ PipeMonitors []*db.PipeMonitor
3132 }
3233
3334 func NewTestDB(logger *slog.Logger) *TestDB {
......@@ -96,6 +97,70 @@ func (t *TestDB) AddPubkey(pubkey *db.PublicKey) {
9697 t.Pubkeys = append(t.Pubkeys, pubkey)
9798 }
9899
100+func (t *TestDB) UpsertPipeMonitor(userID, topic string, dur time.Duration, winEnd *time.Time) error {
101+ for _, m := range t.PipeMonitors {
102+ if m.UserId == userID && m.Topic == topic {
103+ m.WindowDur = dur
104+ m.WindowEnd = winEnd
105+ now := time.Now()
106+ m.UpdatedAt = &now
107+ return nil
108+ }
109+ }
110+ now := time.Now()
111+ t.PipeMonitors = append(t.PipeMonitors, &db.PipeMonitor{
112+ ID: fmt.Sprintf("monitor-%s-%s", userID, topic),
113+ UserId: userID,
114+ Topic: topic,
115+ WindowDur: dur,
116+ WindowEnd: winEnd,
117+ CreatedAt: &now,
118+ UpdatedAt: &now,
119+ })
120+ return nil
121+}
122+
123+func (t *TestDB) UpdatePipeMonitorLastPing(userID, topic string, lastPing *time.Time) error {
124+ for _, m := range t.PipeMonitors {
125+ if m.UserId == userID && m.Topic == topic {
126+ m.LastPing = lastPing
127+ now := time.Now()
128+ m.UpdatedAt = &now
129+ return nil
130+ }
131+ }
132+ return fmt.Errorf("monitor not found")
133+}
134+
135+func (t *TestDB) RemovePipeMonitor(userID, topic string) error {
136+ for i, m := range t.PipeMonitors {
137+ if m.UserId == userID && m.Topic == topic {
138+ t.PipeMonitors = append(t.PipeMonitors[:i], t.PipeMonitors[i+1:]...)
139+ return nil
140+ }
141+ }
142+ return fmt.Errorf("monitor not found")
143+}
144+
145+func (t *TestDB) FindPipeMonitorByTopic(userID, topic string) (*db.PipeMonitor, error) {
146+ for _, m := range t.PipeMonitors {
147+ if m.UserId == userID && m.Topic == topic {
148+ return m, nil
149+ }
150+ }
151+ return nil, fmt.Errorf("monitor not found")
152+}
153+
154+func (t *TestDB) FindPipeMonitorsByUser(userID string) ([]*db.PipeMonitor, error) {
155+ var monitors []*db.PipeMonitor
156+ for _, m := range t.PipeMonitors {
157+ if m.UserId == userID {
158+ monitors = append(monitors, m)
159+ }
160+ }
161+ return monitors, nil
162+}
163+
99164 type TestSSHServer struct {
100165 Cfg *shared.ConfigSite
101166 DBPool *TestDB
......@@ -1393,3 +1458,466 @@ func TestPubSub_MultipleSubscribers(t *testing.T) {
13931458 t.Errorf("subscriber 3 did not receive message, got: %q", string(received3[:n3]))
13941459 }
13951460 }
1461+
1462+// Monitor CLI Tests
1463+
1464+func TestMonitor_UnauthenticatedUserDenied(t *testing.T) {
1465+ server := NewTestSSHServer(t)
1466+ defer server.Shutdown()
1467+
1468+ user := GenerateUser("anonymous")
1469+
1470+ client, err := user.NewClient()
1471+ if err != nil {
1472+ t.Fatalf("failed to connect: %v", err)
1473+ }
1474+ defer func() { _ = client.Close() }()
1475+
1476+ output, err := user.RunCommand(client, "monitor my-service 1h")
1477+ if err != nil {
1478+ t.Logf("command error (expected): %v", err)
1479+ }
1480+
1481+ if !strings.Contains(output, "access denied") {
1482+ t.Errorf("expected 'access denied', got: %s", output)
1483+ }
1484+}
1485+
1486+func TestMonitor_CreateMonitor(t *testing.T) {
1487+ server := NewTestSSHServer(t)
1488+ defer server.Shutdown()
1489+
1490+ user := GenerateUser("alice")
1491+ RegisterUserWithServer(server, user)
1492+
1493+ client, err := user.NewClient()
1494+ if err != nil {
1495+ t.Fatalf("failed to connect: %v", err)
1496+ }
1497+ defer func() { _ = client.Close() }()
1498+
1499+ output, err := user.RunCommand(client, "monitor pico-uptime 24h")
1500+ if err != nil {
1501+ t.Logf("command completed: %v", err)
1502+ }
1503+
1504+ if strings.Contains(output, "access denied") {
1505+ t.Errorf("authenticated user should not get access denied, got: %s", output)
1506+ }
1507+
1508+ // Verify monitor was created in DB
1509+ monitor, err := server.DBPool.FindPipeMonitorByTopic("alice-id", "pico-uptime")
1510+ if err != nil {
1511+ t.Fatalf("monitor should exist in DB: %v", err)
1512+ }
1513+
1514+ if monitor.WindowDur != 24*time.Hour {
1515+ t.Errorf("expected window duration 24h, got: %v", monitor.WindowDur)
1516+ }
1517+
1518+ if !strings.Contains(output, "pico-uptime") || !strings.Contains(output, "24h") {
1519+ t.Errorf("output should confirm monitor creation, got: %s", output)
1520+ }
1521+}
1522+
1523+func TestMonitor_UpdateMonitor(t *testing.T) {
1524+ server := NewTestSSHServer(t)
1525+ defer server.Shutdown()
1526+
1527+ user := GenerateUser("alice")
1528+ RegisterUserWithServer(server, user)
1529+
1530+ client, err := user.NewClient()
1531+ if err != nil {
1532+ t.Fatalf("failed to connect: %v", err)
1533+ }
1534+ defer func() { _ = client.Close() }()
1535+
1536+ // Create initial monitor
1537+ _, err = user.RunCommand(client, "monitor my-cron 1h")
1538+ if err != nil {
1539+ t.Logf("create command completed: %v", err)
1540+ }
1541+
1542+ // Upsert with new duration
1543+ output, err := user.RunCommand(client, "monitor my-cron 6h")
1544+ if err != nil {
1545+ t.Logf("update command completed: %v", err)
1546+ }
1547+
1548+ // Verify monitor was updated
1549+ monitor, err := server.DBPool.FindPipeMonitorByTopic("alice-id", "my-cron")
1550+ if err != nil {
1551+ t.Fatalf("monitor should exist in DB: %v", err)
1552+ }
1553+
1554+ if monitor.WindowDur != 6*time.Hour {
1555+ t.Errorf("expected window duration 6h after update, got: %v", monitor.WindowDur)
1556+ }
1557+
1558+ if !strings.Contains(output, "6h") {
1559+ t.Errorf("output should confirm updated duration, got: %s", output)
1560+ }
1561+}
1562+
1563+func TestMonitor_DeleteMonitor(t *testing.T) {
1564+ server := NewTestSSHServer(t)
1565+ defer server.Shutdown()
1566+
1567+ user := GenerateUser("alice")
1568+ RegisterUserWithServer(server, user)
1569+
1570+ client, err := user.NewClient()
1571+ if err != nil {
1572+ t.Fatalf("failed to connect: %v", err)
1573+ }
1574+ defer func() { _ = client.Close() }()
1575+
1576+ // Create monitor first
1577+ _, err = user.RunCommand(client, "monitor to-delete 1h")
1578+ if err != nil {
1579+ t.Logf("create command completed: %v", err)
1580+ }
1581+
1582+ // Verify it exists
1583+ _, err = server.DBPool.FindPipeMonitorByTopic("alice-id", "to-delete")
1584+ if err != nil {
1585+ t.Fatalf("monitor should exist before deletion: %v", err)
1586+ }
1587+
1588+ // Delete it
1589+ output, err := user.RunCommand(client, "monitor to-delete -d")
1590+ if err != nil {
1591+ t.Logf("delete command completed: %v", err)
1592+ }
1593+
1594+ // Verify it's gone
1595+ _, err = server.DBPool.FindPipeMonitorByTopic("alice-id", "to-delete")
1596+ if err == nil {
1597+ t.Errorf("monitor should be deleted from DB")
1598+ }
1599+
1600+ if !strings.Contains(output, "deleted") && !strings.Contains(output, "removed") {
1601+ t.Logf("output should confirm deletion, got: %s", output)
1602+ }
1603+}
1604+
1605+func TestMonitor_InvalidDuration(t *testing.T) {
1606+ server := NewTestSSHServer(t)
1607+ defer server.Shutdown()
1608+
1609+ user := GenerateUser("alice")
1610+ RegisterUserWithServer(server, user)
1611+
1612+ client, err := user.NewClient()
1613+ if err != nil {
1614+ t.Fatalf("failed to connect: %v", err)
1615+ }
1616+ defer func() { _ = client.Close() }()
1617+
1618+ output, err := user.RunCommand(client, "monitor my-service invaliduration")
1619+ if err != nil {
1620+ t.Logf("command error (expected): %v", err)
1621+ }
1622+
1623+ if !strings.Contains(output, "invalid") && !strings.Contains(output, "duration") && !strings.Contains(output, "error") {
1624+ t.Errorf("expected error about invalid duration, got: %s", output)
1625+ }
1626+}
1627+
1628+func TestMonitor_MissingTopic(t *testing.T) {
1629+ server := NewTestSSHServer(t)
1630+ defer server.Shutdown()
1631+
1632+ user := GenerateUser("alice")
1633+ RegisterUserWithServer(server, user)
1634+
1635+ client, err := user.NewClient()
1636+ if err != nil {
1637+ t.Fatalf("failed to connect: %v", err)
1638+ }
1639+ defer func() { _ = client.Close() }()
1640+
1641+ output, err := user.RunCommand(client, "monitor")
1642+ if err != nil {
1643+ t.Logf("command error (expected): %v", err)
1644+ }
1645+
1646+ // Should show usage or error about missing topic
1647+ if !strings.Contains(output, "Usage") && !strings.Contains(output, "topic") && !strings.Contains(output, "error") {
1648+ t.Errorf("expected usage info or error about missing topic, got: %s", output)
1649+ }
1650+}
1651+
1652+// Status CLI Tests
1653+
1654+func TestStatus_UnauthenticatedUserDenied(t *testing.T) {
1655+ server := NewTestSSHServer(t)
1656+ defer server.Shutdown()
1657+
1658+ user := GenerateUser("anonymous")
1659+
1660+ client, err := user.NewClient()
1661+ if err != nil {
1662+ t.Fatalf("failed to connect: %v", err)
1663+ }
1664+ defer func() { _ = client.Close() }()
1665+
1666+ output, err := user.RunCommand(client, "status")
1667+ if err != nil {
1668+ t.Logf("command error (expected): %v", err)
1669+ }
1670+
1671+ if !strings.Contains(output, "access denied") {
1672+ t.Errorf("expected 'access denied', got: %s", output)
1673+ }
1674+}
1675+
1676+func TestStatus_NoMonitors(t *testing.T) {
1677+ server := NewTestSSHServer(t)
1678+ defer server.Shutdown()
1679+
1680+ user := GenerateUser("alice")
1681+ RegisterUserWithServer(server, user)
1682+
1683+ client, err := user.NewClient()
1684+ if err != nil {
1685+ t.Fatalf("failed to connect: %v", err)
1686+ }
1687+ defer func() { _ = client.Close() }()
1688+
1689+ output, err := user.RunCommand(client, "status")
1690+ if err != nil {
1691+ t.Logf("command completed: %v", err)
1692+ }
1693+
1694+ if !strings.Contains(output, "no monitors") && !strings.Contains(output, "empty") {
1695+ t.Errorf("expected message about no monitors, got: %s", output)
1696+ }
1697+}
1698+
1699+func TestStatus_ShowsMonitorStatus(t *testing.T) {
1700+ server := NewTestSSHServer(t)
1701+ defer server.Shutdown()
1702+
1703+ user := GenerateUser("alice")
1704+ RegisterUserWithServer(server, user)
1705+
1706+ client, err := user.NewClient()
1707+ if err != nil {
1708+ t.Fatalf("failed to connect: %v", err)
1709+ }
1710+ defer func() { _ = client.Close() }()
1711+
1712+ // Create a monitor
1713+ _, err = user.RunCommand(client, "monitor web-check 1h")
1714+ if err != nil {
1715+ t.Logf("create monitor completed: %v", err)
1716+ }
1717+
1718+ // Check status
1719+ output, err := user.RunCommand(client, "status")
1720+ if err != nil {
1721+ t.Logf("status command completed: %v", err)
1722+ }
1723+
1724+ if !strings.Contains(output, "web-check") {
1725+ t.Errorf("status should list the monitor topic, got: %s", output)
1726+ }
1727+}
1728+
1729+func TestStatus_ShowsHealthyUnhealthy(t *testing.T) {
1730+ server := NewTestSSHServer(t)
1731+ defer server.Shutdown()
1732+
1733+ user := GenerateUser("alice")
1734+ RegisterUserWithServer(server, user)
1735+
1736+ // Create monitors directly in DB with different states
1737+ now := time.Now()
1738+ windowEnd := now.Add(1 * time.Hour)
1739+ recentPing := now.Add(-30 * time.Minute) // within window - healthy
1740+ oldPing := now.Add(-2 * time.Hour) // outside window - unhealthy
1741+
1742+ _ = server.DBPool.UpsertPipeMonitor("alice-id", "healthy-service", 1*time.Hour, &windowEnd)
1743+ _ = server.DBPool.UpdatePipeMonitorLastPing("alice-id", "healthy-service", &recentPing)
1744+
1745+ _ = server.DBPool.UpsertPipeMonitor("alice-id", "unhealthy-service", 1*time.Hour, &windowEnd)
1746+ _ = server.DBPool.UpdatePipeMonitorLastPing("alice-id", "unhealthy-service", &oldPing)
1747+
1748+ client, err := user.NewClient()
1749+ if err != nil {
1750+ t.Fatalf("failed to connect: %v", err)
1751+ }
1752+ defer func() { _ = client.Close() }()
1753+
1754+ output, err := user.RunCommand(client, "status")
1755+ if err != nil {
1756+ t.Logf("status command completed: %v", err)
1757+ }
1758+
1759+ if !strings.Contains(output, "healthy-service") {
1760+ t.Errorf("status should list healthy-service, got: %s", output)
1761+ }
1762+
1763+ if !strings.Contains(output, "unhealthy-service") {
1764+ t.Errorf("status should list unhealthy-service, got: %s", output)
1765+ }
1766+
1767+ // Should indicate different health states
1768+ if !strings.Contains(strings.ToLower(output), "healthy") && !strings.Contains(strings.ToLower(output), "ok") && !strings.Contains(output, "✓") {
1769+ t.Logf("status output should indicate health state: %s", output)
1770+ }
1771+}
1772+
1773+// RSS CLI Tests
1774+
1775+func TestRss_UnauthenticatedUserDenied(t *testing.T) {
1776+ server := NewTestSSHServer(t)
1777+ defer server.Shutdown()
1778+
1779+ user := GenerateUser("anonymous")
1780+
1781+ client, err := user.NewClient()
1782+ if err != nil {
1783+ t.Fatalf("failed to connect: %v", err)
1784+ }
1785+ defer func() { _ = client.Close() }()
1786+
1787+ output, err := user.RunCommand(client, "rss")
1788+ if err != nil {
1789+ t.Logf("command error (expected): %v", err)
1790+ }
1791+
1792+ if !strings.Contains(output, "access denied") {
1793+ t.Errorf("expected 'access denied', got: %s", output)
1794+ }
1795+}
1796+
1797+func TestRss_GeneratesValidRSS(t *testing.T) {
1798+ server := NewTestSSHServer(t)
1799+ defer server.Shutdown()
1800+
1801+ user := GenerateUser("alice")
1802+ RegisterUserWithServer(server, user)
1803+
1804+ // Create a monitor
1805+ now := time.Now()
1806+ windowEnd := now.Add(1 * time.Hour)
1807+ _ = server.DBPool.UpsertPipeMonitor("alice-id", "rss-test-service", 1*time.Hour, &windowEnd)
1808+
1809+ client, err := user.NewClient()
1810+ if err != nil {
1811+ t.Fatalf("failed to connect: %v", err)
1812+ }
1813+ defer func() { _ = client.Close() }()
1814+
1815+ output, err := user.RunCommand(client, "rss")
1816+ if err != nil {
1817+ t.Logf("rss command completed: %v", err)
1818+ }
1819+
1820+ // Should output valid RSS XML
1821+ if !strings.Contains(output, "<?xml") || !strings.Contains(output, "<rss") {
1822+ t.Errorf("expected RSS XML output, got: %s", output)
1823+ }
1824+
1825+ if !strings.Contains(output, "rss-test-service") {
1826+ t.Errorf("RSS should contain monitor topic, got: %s", output)
1827+ }
1828+}
1829+
1830+func TestRss_AlertsOnStaleMonitor(t *testing.T) {
1831+ server := NewTestSSHServer(t)
1832+ defer server.Shutdown()
1833+
1834+ user := GenerateUser("alice")
1835+ RegisterUserWithServer(server, user)
1836+
1837+ // Create a stale monitor (last ping outside window)
1838+ now := time.Now()
1839+ windowEnd := now.Add(-30 * time.Minute) // window already ended
1840+ oldPing := now.Add(-2 * time.Hour)
1841+
1842+ _ = server.DBPool.UpsertPipeMonitor("alice-id", "stale-service", 1*time.Hour, &windowEnd)
1843+ _ = server.DBPool.UpdatePipeMonitorLastPing("alice-id", "stale-service", &oldPing)
1844+
1845+ client, err := user.NewClient()
1846+ if err != nil {
1847+ t.Fatalf("failed to connect: %v", err)
1848+ }
1849+ defer func() { _ = client.Close() }()
1850+
1851+ output, err := user.RunCommand(client, "rss")
1852+ if err != nil {
1853+ t.Logf("rss command completed: %v", err)
1854+ }
1855+
1856+ // Should contain alert item for stale service
1857+ if !strings.Contains(output, "stale-service") {
1858+ t.Errorf("RSS should contain stale-service alert, got: %s", output)
1859+ }
1860+
1861+ // Should have item element for the alert
1862+ if !strings.Contains(output, "<item>") {
1863+ t.Errorf("RSS should contain item element for alert, got: %s", output)
1864+ }
1865+}
1866+
1867+// Pub integration with Monitor
1868+
1869+func TestPub_UpdatesMonitorLastPing(t *testing.T) {
1870+ server := NewTestSSHServer(t)
1871+ defer server.Shutdown()
1872+
1873+ user := GenerateUser("alice")
1874+ RegisterUserWithServer(server, user)
1875+
1876+ // Create a monitor first
1877+ now := time.Now()
1878+ windowEnd := now.Add(1 * time.Hour)
1879+ _ = server.DBPool.UpsertPipeMonitor("alice-id", "ping-test", 1*time.Hour, &windowEnd)
1880+
1881+ subClient, err := user.NewClient()
1882+ if err != nil {
1883+ t.Fatalf("failed to connect subscriber: %v", err)
1884+ }
1885+ defer func() { _ = subClient.Close() }()
1886+
1887+ pubClient, err := user.NewClient()
1888+ if err != nil {
1889+ t.Fatalf("failed to connect publisher: %v", err)
1890+ }
1891+ defer func() { _ = pubClient.Close() }()
1892+
1893+ // Start subscriber
1894+ subSession, err := subClient.NewSession()
1895+ if err != nil {
1896+ t.Fatalf("failed to create sub session: %v", err)
1897+ }
1898+ defer func() { _ = subSession.Close() }()
1899+
1900+ if err := subSession.Start("sub ping-test -c"); err != nil {
1901+ t.Fatalf("failed to start sub: %v", err)
1902+ }
1903+
1904+ time.Sleep(100 * time.Millisecond)
1905+
1906+ // Publish to the monitored topic
1907+ _, err = user.RunCommandWithStdin(pubClient, "pub ping-test -c", "health check")
1908+ if err != nil {
1909+ t.Logf("pub command completed: %v", err)
1910+ }
1911+
1912+ // Verify last_ping was updated
1913+ monitor, err := server.DBPool.FindPipeMonitorByTopic("alice-id", "ping-test")
1914+ if err != nil {
1915+ t.Fatalf("monitor should exist: %v", err)
1916+ }
1917+
1918+ if monitor.LastPing == nil {
1919+ t.Errorf("last_ping should be set after pub")
1920+ } else if time.Since(*monitor.LastPing) > 5*time.Second {
1921+ t.Errorf("last_ping should be recent, got: %v", monitor.LastPing)
1922+ }
1923+}
Back to top