dashboard / erock/pico / feat(pipe): add pipe_monitors table #103 rss

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

diff --git a/Makefile b/Makefile
index 946de79..f6a7e51 100644
--- a/Makefile
+++ b/Makefile
@@ -143,10 +143,11 @@ migrate:
 	$(DOCKER_CMD) exec -i $(DB_CONTAINER) psql -U $(PGUSER) -d $(PGDATABASE) < ./sql/migrations/20250410_add_index_analytics_visits_host_list.sql
 	$(DOCKER_CMD) exec -i $(DB_CONTAINER) psql -U $(PGUSER) -d $(PGDATABASE) < ./sql/migrations/20250418_add_project_post_idx_analytics.sql
 	$(DOCKER_CMD) exec -i $(DB_CONTAINER) psql -U $(PGUSER) -d $(PGDATABASE) < ./sql/migrations/20251217_add_access_logs_table.sql
+	$(DOCKER_CMD) exec -i $(DB_CONTAINER) psql -U $(PGUSER) -d $(PGDATABASE) < ./sql/migrations/20251226_add_pipe_monitoring.sql
 .PHONY: migrate
 
 latest:
-	$(DOCKER_CMD) exec -i $(DB_CONTAINER) psql -U $(PGUSER) -d $(PGDATABASE) < ./sql/migrations/20251217_add_access_logs_table.sql
+	$(DOCKER_CMD) exec -i $(DB_CONTAINER) psql -U $(PGUSER) -d $(PGDATABASE) < ./sql/migrations/20251226_add_pipe_monitoring.sql
 .PHONY: latest
 
 psql:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

diff --git a/sql/migrations/20251226_add_pipe_monitoring.sql b/sql/migrations/20251226_add_pipe_monitoring.sql
new file mode 100644
index 0000000..13a65cf
--- /dev/null
+++ b/sql/migrations/20251226_add_pipe_monitoring.sql
@@ -0,0 +1,17 @@
+CREATE TABLE IF NOT EXISTS pipe_monitors (
+  id uuid NOT NULL DEFAULT uuid_generate_v4(),
+  user_id uuid NOT NULL,
+  topic text NOT NULL,
+  window_dur interval NOT NULL,
+  window_end timestamp without time zone NOT NULL DEFAULT NOW(),
+  last_ping timestamp,
+  created_at timestamp without time zone NOT NULL DEFAULT NOW(),
+  updated_at timestamp without time zone NOT NULL DEFAULT NOW(),
+  CONSTRAINT pipe_monitors_unique_topic UNIQUE (user_id, topic),
+  CONSTRAINT pipe_monitoring_pkey PRIMARY KEY (id),
+  CONSTRAINT fk_pipe_monitoring_app_users
+    FOREIGN KEY(user_id)
+  REFERENCES app_users(id)
+  ON DELETE CASCADE
+  ON UPDATE CASCADE
+);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

diff --git a/pkg/db/db.go b/pkg/db/db.go
index 0087cf7..cb7022e 100644
--- a/pkg/db/db.go
+++ b/pkg/db/db.go
@@ -5,6 +5,7 @@ import (
 	"database/sql/driver"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"regexp"
 	"time"
 )
@@ -378,6 +379,36 @@ type TunsEventLog struct {
 	CreatedAt      *time.Time `json:"created_at" db:"created_at"`
 }
 
+type PipeMonitor struct {
+	ID        string        `json:"id" db:"id"`
+	UserId    string        `json:"user_id" db:"user_id"`
+	Topic     string        `json:"topic" db:"topic"`
+	WindowDur time.Duration `json:"window_dur" db:"window_dur"`
+	WindowEnd *time.Time    `json:"window_end" db:"window_end"`
+	LastPing  *time.Time    `json:"last_ping" db:"last_ping"`
+	CreatedAt *time.Time    `json:"created_at" db:"created_at"`
+	UpdatedAt *time.Time    `json:"updated_at" db:"updated_at"`
+}
+
+func (m *PipeMonitor) Status() error {
+	windowStart := m.WindowEnd.Add(-m.WindowDur)
+	lastPingAfterStart := m.LastPing.After(windowStart)
+	if !lastPingAfterStart {
+		return fmt.Errorf("last ping before window start: last_ping=%s window_start=%s", m.LastPing, windowStart)
+	}
+	lastPingBeforeEnd := m.LastPing.Before(*m.WindowEnd)
+	if !lastPingBeforeEnd {
+		// should not happen but just for data validity we add it
+		return fmt.Errorf("last ping after window end: last_ping=%s window_end=%s", m.LastPing, m.WindowEnd)
+	}
+	return nil
+}
+
+func (m *PipeMonitor) GetNextWindow() *time.Time {
+	win := m.WindowEnd.Add(m.WindowDur)
+	return &win
+}
+
 var NameValidator = regexp.MustCompile("^[a-zA-Z0-9]{1,50}$")
 var DenyList = []string{
 	"admin",
@@ -468,5 +499,10 @@ type DB interface {
 	FindPubkeysInAccessLogs(userID string) ([]string, error)
 	FindAccessLogsByPubkey(pubkey string, fromDate *time.Time) ([]*AccessLog, error)
 
+	UpsertPipeMonitor(userID, topic string, dur time.Duration, winEnd *time.Time) error
+	UpdatePipeMonitorLastPing(userID, topic string, lastPing *time.Time) error
+	RemovePipeMonitor(userID, topic string) error
+	FindPipeMonitorByTopic(userID, topic string) (*PipeMonitor, error)
+
 	Close() error
 }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

diff --git a/pkg/db/postgres/storage.go b/pkg/db/postgres/storage.go
index 9c2064c..a83831b 100644
--- a/pkg/db/postgres/storage.go
+++ b/pkg/db/postgres/storage.go
@@ -1514,3 +1514,45 @@ func (me *PsqlDB) InsertAccessLog(log *db.AccessLog) error {
 	)
 	return err
 }
+
+func (me *PsqlDB) UpsertPipeMonitor(userID, topic string, dur time.Duration, winEnd *time.Time) error {
+	durStr := fmt.Sprintf("%d seconds", int64(dur.Seconds()))
+	_, err := me.Db.Exec(
+		`INSERT INTO pipe_monitors (user_id, topic, window_dur, window_end)
+		VALUES ($1, $2, $3::interval, $4)
+		ON CONFLICT (user_id, topic) DO UPDATE SET window_dur = $3::interval, window_end = $4, updated_at = NOW();`,
+		userID,
+		topic,
+		durStr,
+		winEnd,
+	)
+	return err
+}
+
+func (me *PsqlDB) UpdatePipeMonitorLastPing(userID, topic string, lastPing *time.Time) error {
+	_, err := me.Db.Exec(
+		`UPDATE pipe_monitors SET last_ping = $3, updated_at = NOW() WHERE user_id = $1 AND topic = $2;`,
+		userID,
+		topic,
+		lastPing,
+	)
+	return err
+}
+
+func (me *PsqlDB) RemovePipeMonitor(userID, topic string) error {
+	_, err := me.Db.Exec(
+		`DELETE FROM pipe_monitors WHERE user_id = $1 AND topic = $2;`,
+		userID,
+		topic,
+	)
+	return err
+}
+
+func (me *PsqlDB) FindPipeMonitorByTopic(userID, topic string) (*db.PipeMonitor, error) {
+	monitor := &db.PipeMonitor{}
+	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)
+	if err != nil {
+		return nil, err
+	}
+	return monitor, nil
+}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127

diff --git a/pkg/db/postgres/storage_test.go b/pkg/db/postgres/storage_test.go
index 1d12c69..eedd1e8 100644
--- a/pkg/db/postgres/storage_test.go
+++ b/pkg/db/postgres/storage_test.go
@@ -169,7 +169,7 @@ func cleanupTestData(t *testing.T) {
 		"access_logs", "tuns_event_logs", "analytics_visits",
 		"feed_items", "post_aliases", "post_tags", "posts",
 		"projects", "feature_flags", "payment_history", "tokens",
-		"public_keys", "app_users",
+		"public_keys", "pipe_monitors", "app_users",
 	}
 	for _, table := range tables {
 		_, err := testDB.Db.Exec(fmt.Sprintf("DELETE FROM %s", table))
@@ -1472,3 +1472,113 @@ func TestPaymentHistoryData_JSONBRoundtrip(t *testing.T) {
 		t.Errorf("expected tx_id 'tx789', got '%s'", txId)
 	}
 }
+
+// ============ Pipe Monitor Tests ============
+
+func TestUpsertPipeMonitor(t *testing.T) {
+	cleanupTestData(t)
+
+	user, _ := testDB.RegisterUser("pipemonitorowner", "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAI pipemonitorowner", "comment")
+
+	winEnd := time.Now().Add(time.Hour)
+	err := testDB.UpsertPipeMonitor(user.ID, "test-topic", 5*time.Minute, &winEnd)
+	if err != nil {
+		t.Fatalf("UpsertPipeMonitor failed: %v", err)
+	}
+
+	monitor, err := testDB.FindPipeMonitorByTopic(user.ID, "test-topic")
+	if err != nil {
+		t.Fatalf("FindPipeMonitorByTopic failed: %v", err)
+	}
+	if monitor.Topic != "test-topic" {
+		t.Errorf("expected topic 'test-topic', got '%s'", monitor.Topic)
+	}
+	if monitor.WindowDur != 5*time.Minute {
+		t.Errorf("expected window_dur 5m, got %v", monitor.WindowDur)
+	}
+}
+
+func TestUpsertPipeMonitor_Update(t *testing.T) {
+	cleanupTestData(t)
+
+	user, _ := testDB.RegisterUser("pipeupdateowner", "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAI pipeupdateowner", "comment")
+
+	winEnd1 := time.Now().Add(time.Hour)
+	err := testDB.UpsertPipeMonitor(user.ID, "update-topic", 5*time.Minute, &winEnd1)
+	if err != nil {
+		t.Fatalf("first UpsertPipeMonitor failed: %v", err)
+	}
+
+	winEnd2 := time.Now().Add(2 * time.Hour)
+	err = testDB.UpsertPipeMonitor(user.ID, "update-topic", 10*time.Minute, &winEnd2)
+	if err != nil {
+		t.Fatalf("second UpsertPipeMonitor failed: %v", err)
+	}
+
+	monitor, err := testDB.FindPipeMonitorByTopic(user.ID, "update-topic")
+	if err != nil {
+		t.Fatalf("FindPipeMonitorByTopic failed: %v", err)
+	}
+	if monitor.WindowDur != 10*time.Minute {
+		t.Errorf("expected window_dur 10m after update, got %v", monitor.WindowDur)
+	}
+}
+
+func TestUpdatePipeMonitorLastPing(t *testing.T) {
+	cleanupTestData(t)
+
+	user, _ := testDB.RegisterUser("pipepingowner", "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAI pipepingowner", "comment")
+
+	winEnd := time.Now().Add(time.Hour)
+	err := testDB.UpsertPipeMonitor(user.ID, "ping-topic", 5*time.Minute, &winEnd)
+	if err != nil {
+		t.Fatalf("UpsertPipeMonitor failed: %v", err)
+	}
+
+	lastPing := time.Now()
+	err = testDB.UpdatePipeMonitorLastPing(user.ID, "ping-topic", &lastPing)
+	if err != nil {
+		t.Fatalf("UpdatePipeMonitorLastPing failed: %v", err)
+	}
+
+	monitor, err := testDB.FindPipeMonitorByTopic(user.ID, "ping-topic")
+	if err != nil {
+		t.Fatalf("FindPipeMonitorByTopic failed: %v", err)
+	}
+	if monitor.LastPing == nil {
+		t.Error("expected last_ping to be set, got nil")
+	}
+}
+
+func TestRemovePipeMonitor(t *testing.T) {
+	cleanupTestData(t)
+
+	user, _ := testDB.RegisterUser("piperemoveowner", "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAI piperemoveowner", "comment")
+
+	winEnd := time.Now().Add(time.Hour)
+	err := testDB.UpsertPipeMonitor(user.ID, "remove-topic", 5*time.Minute, &winEnd)
+	if err != nil {
+		t.Fatalf("UpsertPipeMonitor failed: %v", err)
+	}
+
+	err = testDB.RemovePipeMonitor(user.ID, "remove-topic")
+	if err != nil {
+		t.Fatalf("RemovePipeMonitor failed: %v", err)
+	}
+
+	_, err = testDB.FindPipeMonitorByTopic(user.ID, "remove-topic")
+	if err == nil {
+		t.Error("expected error finding removed monitor, got nil")
+	}
+}
+
+func TestFindPipeMonitorByTopic_NotFound(t *testing.T) {
+	cleanupTestData(t)
+
+	user, _ := testDB.RegisterUser("pipenotfoundowner", "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAI pipenotfoundowner", "comment")
+
+	_, err := testDB.FindPipeMonitorByTopic(user.ID, "nonexistent-topic")
+	if err == nil {
+		t.Error("expected error for nonexistent monitor, got nil")
+	}
+}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

diff --git a/pkg/db/stub/stub.go b/pkg/db/stub/stub.go
index 3e8e407..1a9d9fc 100644
--- a/pkg/db/stub/stub.go
+++ b/pkg/db/stub/stub.go
@@ -243,3 +243,19 @@ func (me *StubDB) FindPubkeysInAccessLogs(userID string) ([]string, error) {
 func (me *StubDB) InsertAccessLog(log *db.AccessLog) error {
 	return errNotImpl
 }
+
+func (me *StubDB) UpsertPipeMonitor(userID, topic string, dur time.Duration, winEnd *time.Time) error {
+	return errNotImpl
+}
+
+func (me *StubDB) UpdatePipeMonitorLastPing(userID, topic string, lastPing *time.Time) error {
+	return errNotImpl
+}
+
+func (me *StubDB) RemovePipeMonitor(userID, topic string) error {
+	return errNotImpl
+}
+
+func (me *StubDB) FindPipeMonitorByTopic(userID, topic string) (*db.PipeMonitor, error) {
+	return nil, errNotImpl
+}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556

diff --git a/pkg/apps/pipe/ssh_test.go b/pkg/apps/pipe/ssh_test.go
index b13d47b..2f978ed 100644
--- a/pkg/apps/pipe/ssh_test.go
+++ b/pkg/apps/pipe/ssh_test.go
@@ -25,9 +25,10 @@ import (
 
 type TestDB struct {
 	*stub.StubDB
-	Users    []*db.User
-	Pubkeys  []*db.PublicKey
-	Features []*db.FeatureFlag
+	Users        []*db.User
+	Pubkeys      []*db.PublicKey
+	Features     []*db.FeatureFlag
+	PipeMonitors []*db.PipeMonitor
 }
 
 func NewTestDB(logger *slog.Logger) *TestDB {
@@ -96,6 +97,70 @@ func (t *TestDB) AddPubkey(pubkey *db.PublicKey) {
 	t.Pubkeys = append(t.Pubkeys, pubkey)
 }
 
+func (t *TestDB) UpsertPipeMonitor(userID, topic string, dur time.Duration, winEnd *time.Time) error {
+	for _, m := range t.PipeMonitors {
+		if m.UserId == userID && m.Topic == topic {
+			m.WindowDur = dur
+			m.WindowEnd = winEnd
+			now := time.Now()
+			m.UpdatedAt = &now
+			return nil
+		}
+	}
+	now := time.Now()
+	t.PipeMonitors = append(t.PipeMonitors, &db.PipeMonitor{
+		ID:        fmt.Sprintf("monitor-%s-%s", userID, topic),
+		UserId:    userID,
+		Topic:     topic,
+		WindowDur: dur,
+		WindowEnd: winEnd,
+		CreatedAt: &now,
+		UpdatedAt: &now,
+	})
+	return nil
+}
+
+func (t *TestDB) UpdatePipeMonitorLastPing(userID, topic string, lastPing *time.Time) error {
+	for _, m := range t.PipeMonitors {
+		if m.UserId == userID && m.Topic == topic {
+			m.LastPing = lastPing
+			now := time.Now()
+			m.UpdatedAt = &now
+			return nil
+		}
+	}
+	return fmt.Errorf("monitor not found")
+}
+
+func (t *TestDB) RemovePipeMonitor(userID, topic string) error {
+	for i, m := range t.PipeMonitors {
+		if m.UserId == userID && m.Topic == topic {
+			t.PipeMonitors = append(t.PipeMonitors[:i], t.PipeMonitors[i+1:]...)
+			return nil
+		}
+	}
+	return fmt.Errorf("monitor not found")
+}
+
+func (t *TestDB) FindPipeMonitorByTopic(userID, topic string) (*db.PipeMonitor, error) {
+	for _, m := range t.PipeMonitors {
+		if m.UserId == userID && m.Topic == topic {
+			return m, nil
+		}
+	}
+	return nil, fmt.Errorf("monitor not found")
+}
+
+func (t *TestDB) FindPipeMonitorsByUser(userID string) ([]*db.PipeMonitor, error) {
+	var monitors []*db.PipeMonitor
+	for _, m := range t.PipeMonitors {
+		if m.UserId == userID {
+			monitors = append(monitors, m)
+		}
+	}
+	return monitors, nil
+}
+
 type TestSSHServer struct {
 	Cfg    *shared.ConfigSite
 	DBPool *TestDB
@@ -1393,3 +1458,466 @@ func TestPubSub_MultipleSubscribers(t *testing.T) {
 		t.Errorf("subscriber 3 did not receive message, got: %q", string(received3[:n3]))
 	}
 }
+
+// Monitor CLI Tests
+
+func TestMonitor_UnauthenticatedUserDenied(t *testing.T) {
+	server := NewTestSSHServer(t)
+	defer server.Shutdown()
+
+	user := GenerateUser("anonymous")
+
+	client, err := user.NewClient()
+	if err != nil {
+		t.Fatalf("failed to connect: %v", err)
+	}
+	defer func() { _ = client.Close() }()
+
+	output, err := user.RunCommand(client, "monitor my-service 1h")
+	if err != nil {
+		t.Logf("command error (expected): %v", err)
+	}
+
+	if !strings.Contains(output, "access denied") {
+		t.Errorf("expected 'access denied', got: %s", output)
+	}
+}
+
+func TestMonitor_CreateMonitor(t *testing.T) {
+	server := NewTestSSHServer(t)
+	defer server.Shutdown()
+
+	user := GenerateUser("alice")
+	RegisterUserWithServer(server, user)
+
+	client, err := user.NewClient()
+	if err != nil {
+		t.Fatalf("failed to connect: %v", err)
+	}
+	defer func() { _ = client.Close() }()
+
+	output, err := user.RunCommand(client, "monitor pico-uptime 24h")
+	if err != nil {
+		t.Logf("command completed: %v", err)
+	}
+
+	if strings.Contains(output, "access denied") {
+		t.Errorf("authenticated user should not get access denied, got: %s", output)
+	}
+
+	// Verify monitor was created in DB
+	monitor, err := server.DBPool.FindPipeMonitorByTopic("alice-id", "pico-uptime")
+	if err != nil {
+		t.Fatalf("monitor should exist in DB: %v", err)
+	}
+
+	if monitor.WindowDur != 24*time.Hour {
+		t.Errorf("expected window duration 24h, got: %v", monitor.WindowDur)
+	}
+
+	if !strings.Contains(output, "pico-uptime") || !strings.Contains(output, "24h") {
+		t.Errorf("output should confirm monitor creation, got: %s", output)
+	}
+}
+
+func TestMonitor_UpdateMonitor(t *testing.T) {
+	server := NewTestSSHServer(t)
+	defer server.Shutdown()
+
+	user := GenerateUser("alice")
+	RegisterUserWithServer(server, user)
+
+	client, err := user.NewClient()
+	if err != nil {
+		t.Fatalf("failed to connect: %v", err)
+	}
+	defer func() { _ = client.Close() }()
+
+	// Create initial monitor
+	_, err = user.RunCommand(client, "monitor my-cron 1h")
+	if err != nil {
+		t.Logf("create command completed: %v", err)
+	}
+
+	// Upsert with new duration
+	output, err := user.RunCommand(client, "monitor my-cron 6h")
+	if err != nil {
+		t.Logf("update command completed: %v", err)
+	}
+
+	// Verify monitor was updated
+	monitor, err := server.DBPool.FindPipeMonitorByTopic("alice-id", "my-cron")
+	if err != nil {
+		t.Fatalf("monitor should exist in DB: %v", err)
+	}
+
+	if monitor.WindowDur != 6*time.Hour {
+		t.Errorf("expected window duration 6h after update, got: %v", monitor.WindowDur)
+	}
+
+	if !strings.Contains(output, "6h") {
+		t.Errorf("output should confirm updated duration, got: %s", output)
+	}
+}
+
+func TestMonitor_DeleteMonitor(t *testing.T) {
+	server := NewTestSSHServer(t)
+	defer server.Shutdown()
+
+	user := GenerateUser("alice")
+	RegisterUserWithServer(server, user)
+
+	client, err := user.NewClient()
+	if err != nil {
+		t.Fatalf("failed to connect: %v", err)
+	}
+	defer func() { _ = client.Close() }()
+
+	// Create monitor first
+	_, err = user.RunCommand(client, "monitor to-delete 1h")
+	if err != nil {
+		t.Logf("create command completed: %v", err)
+	}
+
+	// Verify it exists
+	_, err = server.DBPool.FindPipeMonitorByTopic("alice-id", "to-delete")
+	if err != nil {
+		t.Fatalf("monitor should exist before deletion: %v", err)
+	}
+
+	// Delete it
+	output, err := user.RunCommand(client, "monitor to-delete -d")
+	if err != nil {
+		t.Logf("delete command completed: %v", err)
+	}
+
+	// Verify it's gone
+	_, err = server.DBPool.FindPipeMonitorByTopic("alice-id", "to-delete")
+	if err == nil {
+		t.Errorf("monitor should be deleted from DB")
+	}
+
+	if !strings.Contains(output, "deleted") && !strings.Contains(output, "removed") {
+		t.Logf("output should confirm deletion, got: %s", output)
+	}
+}
+
+func TestMonitor_InvalidDuration(t *testing.T) {
+	server := NewTestSSHServer(t)
+	defer server.Shutdown()
+
+	user := GenerateUser("alice")
+	RegisterUserWithServer(server, user)
+
+	client, err := user.NewClient()
+	if err != nil {
+		t.Fatalf("failed to connect: %v", err)
+	}
+	defer func() { _ = client.Close() }()
+
+	output, err := user.RunCommand(client, "monitor my-service invaliduration")
+	if err != nil {
+		t.Logf("command error (expected): %v", err)
+	}
+
+	if !strings.Contains(output, "invalid") && !strings.Contains(output, "duration") && !strings.Contains(output, "error") {
+		t.Errorf("expected error about invalid duration, got: %s", output)
+	}
+}
+
+func TestMonitor_MissingTopic(t *testing.T) {
+	server := NewTestSSHServer(t)
+	defer server.Shutdown()
+
+	user := GenerateUser("alice")
+	RegisterUserWithServer(server, user)
+
+	client, err := user.NewClient()
+	if err != nil {
+		t.Fatalf("failed to connect: %v", err)
+	}
+	defer func() { _ = client.Close() }()
+
+	output, err := user.RunCommand(client, "monitor")
+	if err != nil {
+		t.Logf("command error (expected): %v", err)
+	}
+
+	// Should show usage or error about missing topic
+	if !strings.Contains(output, "Usage") && !strings.Contains(output, "topic") && !strings.Contains(output, "error") {
+		t.Errorf("expected usage info or error about missing topic, got: %s", output)
+	}
+}
+
+// Status CLI Tests
+
+func TestStatus_UnauthenticatedUserDenied(t *testing.T) {
+	server := NewTestSSHServer(t)
+	defer server.Shutdown()
+
+	user := GenerateUser("anonymous")
+
+	client, err := user.NewClient()
+	if err != nil {
+		t.Fatalf("failed to connect: %v", err)
+	}
+	defer func() { _ = client.Close() }()
+
+	output, err := user.RunCommand(client, "status")
+	if err != nil {
+		t.Logf("command error (expected): %v", err)
+	}
+
+	if !strings.Contains(output, "access denied") {
+		t.Errorf("expected 'access denied', got: %s", output)
+	}
+}
+
+func TestStatus_NoMonitors(t *testing.T) {
+	server := NewTestSSHServer(t)
+	defer server.Shutdown()
+
+	user := GenerateUser("alice")
+	RegisterUserWithServer(server, user)
+
+	client, err := user.NewClient()
+	if err != nil {
+		t.Fatalf("failed to connect: %v", err)
+	}
+	defer func() { _ = client.Close() }()
+
+	output, err := user.RunCommand(client, "status")
+	if err != nil {
+		t.Logf("command completed: %v", err)
+	}
+
+	if !strings.Contains(output, "no monitors") && !strings.Contains(output, "empty") {
+		t.Errorf("expected message about no monitors, got: %s", output)
+	}
+}
+
+func TestStatus_ShowsMonitorStatus(t *testing.T) {
+	server := NewTestSSHServer(t)
+	defer server.Shutdown()
+
+	user := GenerateUser("alice")
+	RegisterUserWithServer(server, user)
+
+	client, err := user.NewClient()
+	if err != nil {
+		t.Fatalf("failed to connect: %v", err)
+	}
+	defer func() { _ = client.Close() }()
+
+	// Create a monitor
+	_, err = user.RunCommand(client, "monitor web-check 1h")
+	if err != nil {
+		t.Logf("create monitor completed: %v", err)
+	}
+
+	// Check status
+	output, err := user.RunCommand(client, "status")
+	if err != nil {
+		t.Logf("status command completed: %v", err)
+	}
+
+	if !strings.Contains(output, "web-check") {
+		t.Errorf("status should list the monitor topic, got: %s", output)
+	}
+}
+
+func TestStatus_ShowsHealthyUnhealthy(t *testing.T) {
+	server := NewTestSSHServer(t)
+	defer server.Shutdown()
+
+	user := GenerateUser("alice")
+	RegisterUserWithServer(server, user)
+
+	// Create monitors directly in DB with different states
+	now := time.Now()
+	windowEnd := now.Add(1 * time.Hour)
+	recentPing := now.Add(-30 * time.Minute) // within window - healthy
+	oldPing := now.Add(-2 * time.Hour)       // outside window - unhealthy
+
+	_ = server.DBPool.UpsertPipeMonitor("alice-id", "healthy-service", 1*time.Hour, &windowEnd)
+	_ = server.DBPool.UpdatePipeMonitorLastPing("alice-id", "healthy-service", &recentPing)
+
+	_ = server.DBPool.UpsertPipeMonitor("alice-id", "unhealthy-service", 1*time.Hour, &windowEnd)
+	_ = server.DBPool.UpdatePipeMonitorLastPing("alice-id", "unhealthy-service", &oldPing)
+
+	client, err := user.NewClient()
+	if err != nil {
+		t.Fatalf("failed to connect: %v", err)
+	}
+	defer func() { _ = client.Close() }()
+
+	output, err := user.RunCommand(client, "status")
+	if err != nil {
+		t.Logf("status command completed: %v", err)
+	}
+
+	if !strings.Contains(output, "healthy-service") {
+		t.Errorf("status should list healthy-service, got: %s", output)
+	}
+
+	if !strings.Contains(output, "unhealthy-service") {
+		t.Errorf("status should list unhealthy-service, got: %s", output)
+	}
+
+	// Should indicate different health states
+	if !strings.Contains(strings.ToLower(output), "healthy") && !strings.Contains(strings.ToLower(output), "ok") && !strings.Contains(output, "✓") {
+		t.Logf("status output should indicate health state: %s", output)
+	}
+}
+
+// RSS CLI Tests
+
+func TestRss_UnauthenticatedUserDenied(t *testing.T) {
+	server := NewTestSSHServer(t)
+	defer server.Shutdown()
+
+	user := GenerateUser("anonymous")
+
+	client, err := user.NewClient()
+	if err != nil {
+		t.Fatalf("failed to connect: %v", err)
+	}
+	defer func() { _ = client.Close() }()
+
+	output, err := user.RunCommand(client, "rss")
+	if err != nil {
+		t.Logf("command error (expected): %v", err)
+	}
+
+	if !strings.Contains(output, "access denied") {
+		t.Errorf("expected 'access denied', got: %s", output)
+	}
+}
+
+func TestRss_GeneratesValidRSS(t *testing.T) {
+	server := NewTestSSHServer(t)
+	defer server.Shutdown()
+
+	user := GenerateUser("alice")
+	RegisterUserWithServer(server, user)
+
+	// Create a monitor
+	now := time.Now()
+	windowEnd := now.Add(1 * time.Hour)
+	_ = server.DBPool.UpsertPipeMonitor("alice-id", "rss-test-service", 1*time.Hour, &windowEnd)
+
+	client, err := user.NewClient()
+	if err != nil {
+		t.Fatalf("failed to connect: %v", err)
+	}
+	defer func() { _ = client.Close() }()
+
+	output, err := user.RunCommand(client, "rss")
+	if err != nil {
+		t.Logf("rss command completed: %v", err)
+	}
+
+	// Should output valid RSS XML
+	if !strings.Contains(output, "<?xml") || !strings.Contains(output, "<rss") {
+		t.Errorf("expected RSS XML output, got: %s", output)
+	}
+
+	if !strings.Contains(output, "rss-test-service") {
+		t.Errorf("RSS should contain monitor topic, got: %s", output)
+	}
+}
+
+func TestRss_AlertsOnStaleMonitor(t *testing.T) {
+	server := NewTestSSHServer(t)
+	defer server.Shutdown()
+
+	user := GenerateUser("alice")
+	RegisterUserWithServer(server, user)
+
+	// Create a stale monitor (last ping outside window)
+	now := time.Now()
+	windowEnd := now.Add(-30 * time.Minute) // window already ended
+	oldPing := now.Add(-2 * time.Hour)
+
+	_ = server.DBPool.UpsertPipeMonitor("alice-id", "stale-service", 1*time.Hour, &windowEnd)
+	_ = server.DBPool.UpdatePipeMonitorLastPing("alice-id", "stale-service", &oldPing)
+
+	client, err := user.NewClient()
+	if err != nil {
+		t.Fatalf("failed to connect: %v", err)
+	}
+	defer func() { _ = client.Close() }()
+
+	output, err := user.RunCommand(client, "rss")
+	if err != nil {
+		t.Logf("rss command completed: %v", err)
+	}
+
+	// Should contain alert item for stale service
+	if !strings.Contains(output, "stale-service") {
+		t.Errorf("RSS should contain stale-service alert, got: %s", output)
+	}
+
+	// Should have item element for the alert
+	if !strings.Contains(output, "<item>") {
+		t.Errorf("RSS should contain item element for alert, got: %s", output)
+	}
+}
+
+// Pub integration with Monitor
+
+func TestPub_UpdatesMonitorLastPing(t *testing.T) {
+	server := NewTestSSHServer(t)
+	defer server.Shutdown()
+
+	user := GenerateUser("alice")
+	RegisterUserWithServer(server, user)
+
+	// Create a monitor first
+	now := time.Now()
+	windowEnd := now.Add(1 * time.Hour)
+	_ = server.DBPool.UpsertPipeMonitor("alice-id", "ping-test", 1*time.Hour, &windowEnd)
+
+	subClient, err := user.NewClient()
+	if err != nil {
+		t.Fatalf("failed to connect subscriber: %v", err)
+	}
+	defer func() { _ = subClient.Close() }()
+
+	pubClient, err := user.NewClient()
+	if err != nil {
+		t.Fatalf("failed to connect publisher: %v", err)
+	}
+	defer func() { _ = pubClient.Close() }()
+
+	// Start subscriber
+	subSession, err := subClient.NewSession()
+	if err != nil {
+		t.Fatalf("failed to create sub session: %v", err)
+	}
+	defer func() { _ = subSession.Close() }()
+
+	if err := subSession.Start("sub ping-test -c"); err != nil {
+		t.Fatalf("failed to start sub: %v", err)
+	}
+
+	time.Sleep(100 * time.Millisecond)
+
+	// Publish to the monitored topic
+	_, err = user.RunCommandWithStdin(pubClient, "pub ping-test -c", "health check")
+	if err != nil {
+		t.Logf("pub command completed: %v", err)
+	}
+
+	// Verify last_ping was updated
+	monitor, err := server.DBPool.FindPipeMonitorByTopic("alice-id", "ping-test")
+	if err != nil {
+		t.Fatalf("monitor should exist: %v", err)
+	}
+
+	if monitor.LastPing == nil {
+		t.Errorf("last_ping should be set after pub")
+	} else if time.Since(*monitor.LastPing) > 5*time.Second {
+		t.Errorf("last_ping should be recent, got: %v", monitor.LastPing)
+	}
+}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87

diff --git a/pkg/apps/pipe/cli.go b/pkg/apps/pipe/cli.go
index d521fe5..f712e4e 100644
--- a/pkg/apps/pipe/cli.go
+++ b/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
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

diff --git a/pkg/db/db.go b/pkg/db/db.go
index cb7022e..70fd92d 100644
--- a/pkg/db/db.go
+++ b/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
 }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

diff --git a/pkg/db/postgres/storage.go b/pkg/db/postgres/storage.go
index a83831b..f84ad1a 100644
--- a/pkg/db/postgres/storage.go
+++ b/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
+}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

diff --git a/pkg/db/postgres/storage_test.go b/pkg/db/postgres/storage_test.go
index eedd1e8..106d501 100644
--- a/pkg/db/postgres/storage_test.go
+++ b/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))
+	}
+}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

diff --git a/pkg/db/stub/stub.go b/pkg/db/stub/stub.go
index 1a9d9fc..9a67dd3 100644
--- a/pkg/db/stub/stub.go
+++ b/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
+}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139

diff --git a/pkg/apps/pipe/cli.go b/pkg/apps/pipe/cli.go
index f712e4e..e72ca68 100644
--- a/pkg/apps/pipe/cli.go
+++ b/pkg/apps/pipe/cli.go
@@ -9,10 +9,12 @@ import (
 	"log/slog"
 	"slices"
 	"strings"
+	"text/tabwriter"
 	"time"
 
 	"github.com/antoniomika/syncmap"
 	"github.com/google/uuid"
+	"github.com/gorilla/feeds"
 	"github.com/picosh/pico/pkg/db"
 	"github.com/picosh/pico/pkg/pssh"
 	"github.com/picosh/pico/pkg/shared"
@@ -77,6 +79,18 @@ func Middleware(handler *CliHandler) pssh.SSHServerMiddleware {
 					sesh.Fatal(err)
 				}
 				return next(sesh)
+			case "status":
+				err := handler.status(cliCmd, user)
+				if err != nil {
+					sesh.Fatal(err)
+				}
+				return next(sesh)
+			case "rss":
+				err := handler.rss(cliCmd, user)
+				if err != nil {
+					sesh.Fatal(err)
+				}
+				return next(sesh)
 			}
 
 			topic := ""
@@ -343,6 +357,102 @@ func (handler *CliHandler) monitor(cmd *CliCmd, user *db.User) error {
 	return nil
 }
 
+func (handler *CliHandler) status(cmd *CliCmd, user *db.User) error {
+	if user == nil {
+		return fmt.Errorf("access denied")
+	}
+
+	monitors, err := handler.DBPool.FindPipeMonitorsByUser(user.ID)
+	if err != nil {
+		return fmt.Errorf("failed to fetch monitors: %w", err)
+	}
+
+	if len(monitors) == 0 {
+		_, _ = fmt.Fprintln(cmd.sesh, "no monitors found")
+		return nil
+	}
+
+	writer := tabwriter.NewWriter(cmd.sesh, 0, 0, 2, ' ', tabwriter.TabIndent)
+	_, _ = fmt.Fprintln(writer, "Topic\tStatus\tReason\tWindow\tLast Ping\tCreated")
+
+	for _, m := range monitors {
+		status := "healthy"
+		reason := ""
+		if err := m.Status(); err != nil {
+			status = "unhealthy"
+			reason = err.Error()
+		}
+
+		lastPing := "never"
+		if m.LastPing != nil {
+			lastPing = m.LastPing.Format("2006-01-02 15:04:05")
+		}
+
+		createdAt := ""
+		if m.CreatedAt != nil {
+			createdAt = m.CreatedAt.Format("2006-01-02 15:04:05")
+		}
+
+		_, _ = fmt.Fprintf(
+			writer,
+			"%s\t%s\t%s\t%s\t%s\t%s\r\n",
+			m.Topic,
+			status,
+			reason,
+			m.WindowDur.String(),
+			lastPing,
+			createdAt,
+		)
+	}
+	_ = writer.Flush()
+	return nil
+}
+
+func (handler *CliHandler) rss(cmd *CliCmd, user *db.User) error {
+	if user == nil {
+		return fmt.Errorf("access denied")
+	}
+
+	monitors, err := handler.DBPool.FindPipeMonitorsByUser(user.ID)
+	if err != nil {
+		return fmt.Errorf("failed to fetch monitors: %w", err)
+	}
+
+	now := time.Now()
+	feed := &feeds.Feed{
+		Title:       fmt.Sprintf("Pipe Monitors for %s", user.Name),
+		Link:        &feeds.Link{Href: fmt.Sprintf("https://%s", handler.Cfg.Domain)},
+		Description: "Alerts for pipe monitor status changes",
+		Author:      &feeds.Author{Name: user.Name},
+		Created:     now,
+	}
+
+	var feedItems []*feeds.Item
+	for _, m := range monitors {
+		if err := m.Status(); err != nil {
+			item := &feeds.Item{
+				Id:          fmt.Sprintf("%s-%s-%d", user.ID, m.Topic, now.Unix()),
+				Title:       fmt.Sprintf("ALERT: %s is unhealthy", m.Topic),
+				Link:        &feeds.Link{Href: fmt.Sprintf("https://%s", handler.Cfg.Domain)},
+				Description: err.Error(),
+				Created:     now,
+				Updated:     now,
+				Author:      &feeds.Author{Name: user.Name},
+			}
+			feedItems = append(feedItems, item)
+		}
+	}
+	feed.Items = feedItems
+
+	rss, err := feed.ToRss()
+	if err != nil {
+		return fmt.Errorf("failed to generate RSS: %w", err)
+	}
+
+	_, _ = fmt.Fprint(cmd.sesh, rss)
+	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
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

diff --git a/pkg/db/db.go b/pkg/db/db.go
index 70fd92d..48ad9c3 100644
--- a/pkg/db/db.go
+++ b/pkg/db/db.go
@@ -391,6 +391,12 @@ type PipeMonitor struct {
 }
 
 func (m *PipeMonitor) Status() error {
+	if m.LastPing == nil {
+		return fmt.Errorf("no ping received yet")
+	}
+	if m.WindowEnd == nil {
+		return fmt.Errorf("window end not set")
+	}
 	windowStart := m.WindowEnd.Add(-m.WindowDur)
 	lastPingAfterStart := m.LastPing.After(windowStart)
 	if !lastPingAfterStart {

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104

diff --git a/pkg/apps/pipe/cli.go b/pkg/apps/pipe/cli.go
index e72ca68..ee3e2bb 100644
--- a/pkg/apps/pipe/cli.go
+++ b/pkg/apps/pipe/cli.go
@@ -60,6 +60,7 @@ func Middleware(handler *CliHandler) pssh.SSHServerMiddleware {
 				isAdmin:  isAdmin,
 				pipeCtx:  pipeCtx,
 				cancel:   cancel,
+				user:     user,
 			}
 
 			cmd := strings.TrimSpace(args[0])
@@ -181,6 +182,7 @@ type CliCmd struct {
 	isAdmin  bool
 	pipeCtx  context.Context
 	cancel   context.CancelFunc
+	user     *db.User
 }
 
 func help(cfg *shared.ConfigSite, sesh *pssh.SSHServerConnSession) {
@@ -320,12 +322,21 @@ func (handler *CliHandler) monitor(cmd *CliCmd, user *db.User) error {
 		return fmt.Errorf("topic is required")
 	}
 
+	// Resolve to fully qualified topic name
+	result := resolveTopic(TopicResolveInput{
+		UserName: cmd.userName,
+		Topic:    topic,
+		IsAdmin:  cmd.isAdmin,
+		IsPublic: false,
+	})
+	resolvedTopic := result.Name
+
 	if *del {
-		err := handler.DBPool.RemovePipeMonitor(user.ID, topic)
+		err := handler.DBPool.RemovePipeMonitor(user.ID, resolvedTopic)
 		if err != nil {
 			return fmt.Errorf("failed to delete monitor: %w", err)
 		}
-		_, _ = fmt.Fprintf(cmd.sesh, "monitor deleted: %s\r\n", topic)
+		_, _ = fmt.Fprintf(cmd.sesh, "monitor deleted: %s\r\n", resolvedTopic)
 		return nil
 	}
 
@@ -348,12 +359,12 @@ func (handler *CliHandler) monitor(cmd *CliCmd, user *db.User) error {
 	}
 
 	winEnd := time.Now().Add(dur)
-	err = handler.DBPool.UpsertPipeMonitor(user.ID, topic, dur, &winEnd)
+	err = handler.DBPool.UpsertPipeMonitor(user.ID, resolvedTopic, 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)
+	_, _ = fmt.Fprintf(cmd.sesh, "monitor created: %s (window: %s)\r\n", resolvedTopic, dur)
 	return nil
 }
 
@@ -672,9 +683,35 @@ func (handler *CliHandler) pub(cmd *CliCmd, topic string, clientID string) error
 		return err
 	}
 
+	handler.updateMonitor(cmd, name)
+
 	return nil
 }
 
+func (handler *CliHandler) updateMonitor(cmd *CliCmd, topic string) {
+	if cmd.user == nil {
+		return
+	}
+
+	monitor, err := handler.DBPool.FindPipeMonitorByTopic(cmd.user.ID, topic)
+	if err != nil || monitor == nil {
+		return
+	}
+
+	now := time.Now()
+	if err := handler.DBPool.UpdatePipeMonitorLastPing(cmd.user.ID, topic, &now); err != nil {
+		handler.Logger.Error("failed to update monitor last_ping", "err", err, "topic", topic)
+	}
+
+	// Advance window if current time is past window_end
+	if monitor.WindowEnd != nil && now.After(*monitor.WindowEnd) {
+		nextWindow := monitor.GetNextWindow()
+		if err := handler.DBPool.UpsertPipeMonitor(cmd.user.ID, topic, monitor.WindowDur, nextWindow); err != nil {
+			handler.Logger.Error("failed to advance monitor window", "err", err, "topic", topic)
+		}
+	}
+}
+
 func (handler *CliHandler) sub(cmd *CliCmd, topic string, clientID string) error {
 	subCmd := flagSet("sub", cmd.sesh)
 	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
 		return writeErr
 	}
 
+	handler.updateMonitor(cmd, name)
+
 	return nil
 }
 

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81