Commit b2aa9b3

Eric Bower  ·  2026-01-08 15:43:06 -0500 EST
parent 2c02fb4 
feat(pipe): monitors

This change introduces pipe monitors: a way to receive status updates on
your pipes.
8 files changed,  +1686, -20
M Makefile
M pkg/apps/pipe/cli.go
M pkg/apps/pipe/ssh_test.go
M pkg/db/db.go
M pkg/db/postgres/storage.go
M pkg/db/postgres/storage_test.go
M pkg/db/stub/stub.go
A sql/migrations/20251226_add_pipe_monitoring.sql
M Makefile
+2, -1
......@@ -143,10 +143,11 @@ migrate:
143143 $(DOCKER_CMD) exec -i $(DB_CONTAINER) psql -U $(PGUSER) -d $(PGDATABASE) < ./sql/migrations/20250410_add_index_analytics_visits_host_list.sql
144144 $(DOCKER_CMD) exec -i $(DB_CONTAINER) psql -U $(PGUSER) -d $(PGDATABASE) < ./sql/migrations/20250418_add_project_post_idx_analytics.sql
145145 $(DOCKER_CMD) exec -i $(DB_CONTAINER) psql -U $(PGUSER) -d $(PGDATABASE) < ./sql/migrations/20251217_add_access_logs_table.sql
146+ $(DOCKER_CMD) exec -i $(DB_CONTAINER) psql -U $(PGUSER) -d $(PGDATABASE) < ./sql/migrations/20251226_add_pipe_monitoring.sql
146147 .PHONY: migrate
147148
148149 latest:
149- $(DOCKER_CMD) exec -i $(DB_CONTAINER) psql -U $(PGUSER) -d $(PGDATABASE) < ./sql/migrations/20251217_add_access_logs_table.sql
150+ $(DOCKER_CMD) exec -i $(DB_CONTAINER) psql -U $(PGUSER) -d $(PGDATABASE) < ./sql/migrations/20251226_add_pipe_monitoring.sql
150151 .PHONY: latest
151152
152153 psql:
M pkg/apps/pipe/cli.go
+432, -9
......@@ -3,16 +3,21 @@ package pipe
33 import (
44 "bytes"
55 "context"
6+ "database/sql"
7+ "errors"
68 "flag"
79 "fmt"
810 "io"
911 "log/slog"
1012 "slices"
1113 "strings"
14+ "sync/atomic"
15+ "text/tabwriter"
1216 "time"
1317
1418 "github.com/antoniomika/syncmap"
1519 "github.com/google/uuid"
20+ "github.com/gorilla/feeds"
1621 "github.com/picosh/pico/pkg/db"
1722 "github.com/picosh/pico/pkg/pssh"
1823 "github.com/picosh/pico/pkg/shared"
......@@ -58,6 +63,7 @@ func Middleware(handler *CliHandler) pssh.SSHServerMiddleware {
5863 isAdmin: isAdmin,
5964 pipeCtx: pipeCtx,
6065 cancel: cancel,
66+ user: user,
6167 }
6268
6369 cmd := strings.TrimSpace(args[0])
......@@ -68,6 +74,28 @@ func Middleware(handler *CliHandler) pssh.SSHServerMiddleware {
6874 case "ls":
6975 err := handler.ls(cliCmd)
7076 if err != nil {
77+ logger.Error("ls cmd", "err", err)
78+ sesh.Fatal(err)
79+ }
80+ return next(sesh)
81+ case "monitor":
82+ err := handler.monitor(cliCmd, user)
83+ if err != nil {
84+ logger.Error("monitor cmd", "err", err)
85+ sesh.Fatal(err)
86+ }
87+ return next(sesh)
88+ case "status":
89+ err := handler.status(cliCmd, user)
90+ if err != nil {
91+ logger.Error("status cmd", "err", err)
92+ sesh.Fatal(err)
93+ }
94+ return next(sesh)
95+ case "rss":
96+ err := handler.rss(cliCmd, user)
97+ if err != nil {
98+ logger.Error("rss cmd", "err", err)
7199 sesh.Fatal(err)
72100 }
73101 return next(sesh)
......@@ -122,16 +150,25 @@ func Middleware(handler *CliHandler) pssh.SSHServerMiddleware {
122150 case "pub":
123151 err := handler.pub(cliCmd, topic, clientID)
124152 if err != nil {
153+ logger.Error("pub cmd", "err", err)
125154 sesh.Fatal(err)
126155 }
127156 case "sub":
128157 err := handler.sub(cliCmd, topic, clientID)
129158 if err != nil {
159+ logger.Error("sub cmd", "err", err)
130160 sesh.Fatal(err)
131161 }
132162 case "pipe":
133163 err := handler.pipe(cliCmd, topic, clientID)
134164 if err != nil {
165+ logger.Error("pipe cmd", "err", err)
166+ sesh.Fatal(err)
167+ }
168+ case "uptime":
169+ err := handler.uptime(cliCmd, topic, user)
170+ if err != nil {
171+ logger.Error("uptime cmd", "err", err)
135172 sesh.Fatal(err)
136173 }
137174 }
......@@ -161,10 +198,11 @@ type CliCmd struct {
161198 isAdmin bool
162199 pipeCtx context.Context
163200 cancel context.CancelFunc
201+ user *db.User
164202 }
165203
166204 func help(cfg *shared.ConfigSite, sesh *pssh.SSHServerConnSession) {
167- data := fmt.Sprintf(`Command: ssh %s <help | ls | pub | sub | pipe> <topic> [-h | args...]
205+ data := fmt.Sprintf(`Command: ssh %s <command> [args...]
168206
169207 The simplest authenticated pubsub system. Send messages through
170208 user-defined topics. Topics are private to the authenticated
......@@ -175,13 +213,22 @@ at least one event to be sent or received. Pipe ("pipe") allows
175213 for bidirectional messages to be sent between any clients connected
176214 to a pipe.
177215
178-Think of these different commands in terms of the direction the
179-data is being sent:
216+Commands:
217+ help Show this help message
218+ ls List active pubsub channels
219+ pub <topic> [flags] Publish messages to a topic
220+ sub <topic> [flags] Subscribe to messages from a topic
221+ pipe <topic> [flags] Bidirectional messaging between clients
222+
223+Monitoring commands:
224+ monitor <topic> <duration> Create/update a health monitor for a topic
225+ monitor <topic> -d Delete a monitor
226+ status Show health status of all monitors
227+ uptime Show uptime for a topic
228+ rss Get RSS feed of monitor alerts
180229
181-- pub => writes to client
182-- sub => reads from client
183-- pipe => read and write between clients
184-`, toSshCmd(cfg))
230+Use "ssh %s <command> -h" for help on a specific command.
231+`, toSshCmd(cfg), toSshCmd(cfg))
185232
186233 data = strings.ReplaceAll(data, "\n", "\r\n")
187234 _, _ = fmt.Fprintln(sesh, data)
......@@ -274,6 +321,272 @@ func (handler *CliHandler) ls(cmd *CliCmd) error {
274321 return nil
275322 }
276323
324+func (handler *CliHandler) monitor(cmd *CliCmd, user *db.User) error {
325+ if user == nil {
326+ return fmt.Errorf("access denied")
327+ }
328+
329+ args := cmd.sesh.Command()
330+ topic := ""
331+ cmdArgs := args[1:]
332+ if len(args) > 1 && !strings.HasPrefix(args[1], "-") {
333+ topic = strings.TrimSpace(args[1])
334+ cmdArgs = args[2:]
335+ }
336+
337+ monitorCmd := flagSet("monitor", cmd.sesh)
338+ del := monitorCmd.Bool("d", false, "Delete the monitor")
339+
340+ if !flagCheck(monitorCmd, topic, cmdArgs) {
341+ return nil
342+ }
343+
344+ if topic == "" {
345+ _, _ = fmt.Fprintln(cmd.sesh, "Usage: monitor <topic> <duration>")
346+ _, _ = fmt.Fprintln(cmd.sesh, " monitor <topic> -d")
347+ return fmt.Errorf("topic is required")
348+ }
349+
350+ // Resolve to fully qualified topic name
351+ result := resolveTopic(TopicResolveInput{
352+ UserName: cmd.userName,
353+ Topic: topic,
354+ IsAdmin: cmd.isAdmin,
355+ IsPublic: false,
356+ })
357+ resolvedTopic := result.Name
358+
359+ if *del {
360+ handler.Logger.Info("removing pipe monitor", "topic", resolvedTopic)
361+ err := handler.DBPool.RemovePipeMonitor(user.ID, resolvedTopic)
362+ if err != nil {
363+ return fmt.Errorf("failed to delete monitor: %w", err)
364+ }
365+ _, _ = fmt.Fprintf(cmd.sesh, "monitor deleted: %s\r\n", resolvedTopic)
366+ return nil
367+ }
368+
369+ // Create/update monitor - need duration argument
370+ durStr := ""
371+ if monitorCmd.NArg() > 0 {
372+ durStr = monitorCmd.Arg(0)
373+ } else if len(cmdArgs) > 0 {
374+ durStr = cmdArgs[0]
375+ }
376+
377+ if durStr == "" {
378+ _, _ = fmt.Fprintln(cmd.sesh, "Usage: monitor <topic> <duration>")
379+ return fmt.Errorf("duration is required")
380+ }
381+
382+ dur, err := time.ParseDuration(durStr)
383+ if err != nil {
384+ return fmt.Errorf("invalid duration %q: %w", durStr, err)
385+ }
386+
387+ winEnd := time.Now().UTC().Add(dur)
388+ handler.Logger.Info(
389+ "upserting pipe monitor",
390+ "topic", resolvedTopic,
391+ "dur", dur,
392+ "window", winEnd.UTC().Format(time.RFC3339),
393+ )
394+ err = handler.DBPool.UpsertPipeMonitor(user.ID, resolvedTopic, dur, &winEnd)
395+ if err != nil {
396+ return fmt.Errorf("failed to create monitor: %w", err)
397+ }
398+
399+ _, _ = fmt.Fprintf(cmd.sesh, "monitor created: %s (window: %s)\r\n", resolvedTopic, dur)
400+ return nil
401+}
402+
403+func (handler *CliHandler) status(cmd *CliCmd, user *db.User) error {
404+ if user == nil {
405+ return fmt.Errorf("access denied")
406+ }
407+
408+ monitors, err := handler.DBPool.FindPipeMonitorsByUser(user.ID)
409+ if err != nil {
410+ return fmt.Errorf("failed to fetch monitors: %w", err)
411+ }
412+
413+ if len(monitors) == 0 {
414+ _, _ = fmt.Fprintln(cmd.sesh, "no monitors found")
415+ return nil
416+ }
417+
418+ writer := tabwriter.NewWriter(cmd.sesh, 0, 0, 2, ' ', tabwriter.TabIndent)
419+ _, _ = fmt.Fprintln(writer, "Topic\tStatus\tWindow\tLast Ping\tWindow End\tReason")
420+
421+ for _, m := range monitors {
422+ status := "healthy"
423+ reason := ""
424+ if err := m.Status(); err != nil {
425+ status = "unhealthy"
426+ reason = err.Error()
427+ }
428+
429+ lastPing := "never"
430+ if m.LastPing != nil {
431+ lastPing = m.LastPing.UTC().Format(time.RFC3339)
432+ }
433+
434+ windowEnd := ""
435+ if m.WindowEnd != nil {
436+ windowEnd = m.WindowEnd.UTC().Format(time.RFC3339)
437+ }
438+
439+ _, _ = fmt.Fprintf(
440+ writer,
441+ "%s\t%s\t%s\t%s\t%s\t%s\r\n",
442+ m.Topic,
443+ status,
444+ m.WindowDur.String(),
445+ lastPing,
446+ windowEnd,
447+ reason,
448+ )
449+ }
450+ _ = writer.Flush()
451+ return nil
452+}
453+
454+func (handler *CliHandler) uptime(cmd *CliCmd, topic string, user *db.User) error {
455+ if user == nil {
456+ return fmt.Errorf("access denied")
457+ }
458+
459+ if topic == "" {
460+ _, _ = fmt.Fprintln(cmd.sesh, "usage: uptime <topic> [--from <time>] [--to <time>]")
461+ _, _ = fmt.Fprintln(cmd.sesh, " --from: start time (RFC3339 or duration like '24h', '7d', default: 24h)")
462+ _, _ = fmt.Fprintln(cmd.sesh, " --to: end time (RFC3339, default: now)")
463+ return nil
464+ }
465+
466+ fs := flag.NewFlagSet("uptime", flag.ContinueOnError)
467+ fs.SetOutput(cmd.sesh)
468+ fromStr := fs.String("from", "", "start time (RFC3339 or duration like '24h', '7d')")
469+ toStr := fs.String("to", "", "end time (RFC3339, defaults to now)")
470+
471+ if err := fs.Parse(cmd.args); err != nil {
472+ return nil
473+ }
474+
475+ topicResult := resolveTopic(TopicResolveInput{
476+ UserName: cmd.userName,
477+ Topic: topic,
478+ IsAdmin: cmd.isAdmin,
479+ IsPublic: false,
480+ })
481+ resolvedTopic := topicResult.Name
482+
483+ monitor, err := handler.DBPool.FindPipeMonitorByTopic(user.ID, resolvedTopic)
484+ if err != nil {
485+ if errors.Is(err, sql.ErrNoRows) {
486+ return fmt.Errorf("monitor not found: %s", topic)
487+ }
488+ return fmt.Errorf("failed to find monitor: %w", err)
489+ }
490+
491+ now := time.Now().UTC()
492+ to := now
493+ from := now.Add(-24 * time.Hour)
494+
495+ if *fromStr != "" {
496+ if parsed, err := time.Parse(time.RFC3339, *fromStr); err == nil {
497+ from = parsed.UTC()
498+ } else if dur, err := parseDuration(*fromStr); err == nil {
499+ from = now.Add(-dur)
500+ } else {
501+ return fmt.Errorf("invalid --from value: %s", *fromStr)
502+ }
503+ }
504+
505+ if *toStr != "" {
506+ if parsed, err := time.Parse(time.RFC3339, *toStr); err == nil {
507+ to = parsed.UTC()
508+ } else {
509+ return fmt.Errorf("invalid --to value: %s", *toStr)
510+ }
511+ }
512+
513+ history, err := handler.DBPool.FindPipeMonitorHistory(monitor.ID, from, to)
514+ if err != nil {
515+ return fmt.Errorf("failed to fetch history: %w", err)
516+ }
517+
518+ result := db.ComputeUptime(history, from, to)
519+
520+ _, _ = fmt.Fprintf(cmd.sesh, "Monitor: %s\r\n", topic)
521+ _, _ = fmt.Fprintf(cmd.sesh, "Period: %s to %s\r\n", from.Format(time.RFC3339), to.Format(time.RFC3339))
522+ _, _ = fmt.Fprintf(cmd.sesh, "Total Duration: %s\r\n", result.TotalDuration.Round(time.Second))
523+ _, _ = fmt.Fprintf(cmd.sesh, "Uptime Duration: %s\r\n", result.UptimeDuration.Round(time.Second))
524+ _, _ = fmt.Fprintf(cmd.sesh, "Uptime: %.2f%%\r\n", result.UptimePercent)
525+
526+ return nil
527+}
528+
529+func parseDuration(s string) (time.Duration, error) {
530+ if len(s) == 0 {
531+ return 0, fmt.Errorf("empty duration")
532+ }
533+ last := s[len(s)-1]
534+ if last == 'd' {
535+ var n int
536+ _, err := fmt.Sscanf(s, "%d", &n)
537+ if err != nil {
538+ return 0, fmt.Errorf("invalid duration: %s", s)
539+ }
540+ return time.Duration(n) * 24 * time.Hour, nil
541+ }
542+ return time.ParseDuration(s)
543+}
544+
545+func (handler *CliHandler) rss(cmd *CliCmd, user *db.User) error {
546+ if user == nil {
547+ return fmt.Errorf("access denied")
548+ }
549+
550+ monitors, err := handler.DBPool.FindPipeMonitorsByUser(user.ID)
551+ if err != nil {
552+ return fmt.Errorf("failed to fetch monitors: %w", err)
553+ }
554+
555+ now := time.Now()
556+ feed := &feeds.Feed{
557+ Title: fmt.Sprintf("Pipe Monitors for %s", user.Name),
558+ Link: &feeds.Link{Href: fmt.Sprintf("https://%s", handler.Cfg.Domain)},
559+ Description: "Alerts for pipe monitor status changes",
560+ Author: &feeds.Author{Name: user.Name},
561+ Created: now,
562+ }
563+
564+ var feedItems []*feeds.Item
565+ for _, m := range monitors {
566+ if err := m.Status(); err != nil {
567+ item := &feeds.Item{
568+ Id: fmt.Sprintf("%s-%s-%d", user.ID, m.Topic, now.Unix()),
569+ Title: fmt.Sprintf("ALERT: %s is unhealthy", m.Topic),
570+ Link: &feeds.Link{Href: fmt.Sprintf("https://%s", handler.Cfg.Domain)},
571+ Description: err.Error(),
572+ Created: now,
573+ Updated: now,
574+ Author: &feeds.Author{Name: user.Name},
575+ }
576+ feedItems = append(feedItems, item)
577+ }
578+ }
579+ feed.Items = feedItems
580+
581+ rss, err := feed.ToRss()
582+ if err != nil {
583+ return fmt.Errorf("failed to generate RSS: %w", err)
584+ }
585+
586+ _, _ = fmt.Fprint(cmd.sesh, rss)
587+ return nil
588+}
589+
277590 func (handler *CliHandler) pub(cmd *CliCmd, topic string, clientID string) error {
278591 pubCmd := flagSet("pub", cmd.sesh)
279592 access := pubCmd.String("a", "", "Comma separated list of pico usernames or ssh-key fingerprints to allow access to a topic")
......@@ -475,10 +788,12 @@ func (handler *CliHandler) pub(cmd *CliCmd, topic string, clientID string) error
475788 _, _ = fmt.Fprintln(cmd.sesh, "sending msg ...")
476789 }
477790
791+ throttledRW := newThrottledMonitorRW(rw, handler, cmd, name)
792+
478793 err := handler.PubSub.Pub(
479794 cmd.pipeCtx,
480795 clientID,
481- rw,
796+ throttledRW,
482797 []*psub.Channel{
483798 psub.NewChannel(name),
484799 },
......@@ -493,9 +808,113 @@ func (handler *CliHandler) pub(cmd *CliCmd, topic string, clientID string) error
493808 return err
494809 }
495810
811+ handler.updateMonitor(cmd, name)
812+
496813 return nil
497814 }
498815
816+func (handler *CliHandler) updateMonitor(cmd *CliCmd, topic string) {
817+ if cmd.user == nil {
818+ return
819+ }
820+
821+ handler.Logger.Info("update monitor", "topic", topic)
822+ monitor, err := handler.DBPool.FindPipeMonitorByTopic(cmd.user.ID, topic)
823+ if err != nil || monitor == nil {
824+ handler.Logger.Info("no monitor found", "topic", topic)
825+ return
826+ }
827+
828+ now := time.Now().UTC()
829+
830+ // Fixed window semantics: windows are discrete, non-overlapping time slots.
831+ // - last_ping: always updated to show most recent activity (user visibility)
832+ // - window_end: only advances when current time exceeds it (health scheduling)
833+
834+ // If we're past the current window, advance to the window containing `now`
835+ newWindowEnd := *monitor.WindowEnd
836+ if !now.Before(*monitor.WindowEnd) {
837+ // Record history for the completed window before advancing
838+ // This captures that the old window was healthy (had activity)
839+ if err := handler.DBPool.InsertPipeMonitorHistory(monitor.ID, monitor.WindowDur, monitor.WindowEnd, monitor.LastPing); err != nil {
840+ handler.Logger.Error("failed to insert monitor history", "err", err, "topic", topic)
841+ }
842+
843+ // Calculate which window period `now` falls into
844+ elapsed := now.Sub(*monitor.WindowEnd)
845+ periods := int(elapsed/monitor.WindowDur) + 1
846+ newWindowEnd = monitor.WindowEnd.Add(time.Duration(periods) * monitor.WindowDur)
847+
848+ if err := handler.DBPool.UpsertPipeMonitor(cmd.user.ID, topic, monitor.WindowDur, &newWindowEnd); err != nil {
849+ handler.Logger.Error("failed to advance monitor window", "err", err, "topic", topic)
850+ }
851+ handler.Logger.Info("advanced monitor window",
852+ "topic", topic,
853+ "oldWindowEnd", monitor.WindowEnd.Format(time.RFC3339),
854+ "newWindowEnd", newWindowEnd.Format(time.RFC3339),
855+ "periodsMissed", periods-1,
856+ )
857+ }
858+
859+ // Always record the latest ping for user visibility
860+ if err := handler.DBPool.UpdatePipeMonitorLastPing(cmd.user.ID, topic, &now); err != nil {
861+ handler.Logger.Error("failed to update monitor last_ping", "err", err, "topic", topic)
862+ }
863+
864+ handler.Logger.Info("recorded monitor ping",
865+ "topic", topic,
866+ "pingTime", now.Format(time.RFC3339),
867+ "windowEnd", newWindowEnd.Format(time.RFC3339),
868+ )
869+}
870+
871+const monitorThrottleInterval = 15 * time.Second
872+
873+type throttledMonitorRW struct {
874+ rw io.ReadWriter
875+ handler *CliHandler
876+ cmd *CliCmd
877+ topic string
878+ lastPing atomic.Int64 // Unix nanoseconds
879+}
880+
881+func newThrottledMonitorRW(rw io.ReadWriter, handler *CliHandler, cmd *CliCmd, topic string) *throttledMonitorRW {
882+ return &throttledMonitorRW{
883+ rw: rw,
884+ handler: handler,
885+ cmd: cmd,
886+ topic: topic,
887+ }
888+}
889+
890+func (t *throttledMonitorRW) throttledUpdate() {
891+ now := time.Now().UnixNano()
892+ last := t.lastPing.Load()
893+
894+ // First ping (last == 0) or interval elapsed
895+ if last == 0 || now-last >= int64(monitorThrottleInterval) {
896+ if t.lastPing.CompareAndSwap(last, now) {
897+ t.handler.updateMonitor(t.cmd, t.topic)
898+ }
899+ }
900+}
901+
902+func (t *throttledMonitorRW) Read(p []byte) (int, error) {
903+ n, err := t.rw.Read(p)
904+ if n > 0 {
905+ t.throttledUpdate()
906+ }
907+ return n, err
908+}
909+
910+func (t *throttledMonitorRW) Write(p []byte) (int, error) {
911+ n, err := t.rw.Write(p)
912+ if n > 0 {
913+ t.throttledUpdate()
914+ }
915+ return n, err
916+}
917+
499918 func (handler *CliHandler) sub(cmd *CliCmd, topic string, clientID string) error {
500919 subCmd := flagSet("sub", cmd.sesh)
501920 access := subCmd.String("a", "", "Comma separated list of pico usernames or ssh-key fingerprints to allow access to a topic")
......@@ -675,10 +1094,12 @@ func (handler *CliHandler) pipe(cmd *CliCmd, topic string, clientID string) erro
6751094 )
6761095 }
6771096
1097+ throttledRW := newThrottledMonitorRW(cmd.sesh, handler, cmd, name)
1098+
6781099 readErr, writeErr := handler.PubSub.Pipe(
6791100 cmd.pipeCtx,
6801101 clientID,
681- cmd.sesh,
1102+ throttledRW,
6821103 []*psub.Channel{
6831104 psub.NewChannel(name),
6841105 },
......@@ -693,6 +1114,8 @@ func (handler *CliHandler) pipe(cmd *CliCmd, topic string, clientID string) erro
6931114 return writeErr
6941115 }
6951116
1117+ handler.updateMonitor(cmd, name)
1118+
6961119 return nil
6971120 }
6981121
M pkg/apps/pipe/ssh_test.go
+817, -9
......@@ -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,10 +97,83 @@ 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+
164+func (t *TestDB) InsertPipeMonitorHistory(monitorID string, windowDur time.Duration, windowEnd, lastPing *time.Time) error {
165+ return nil
166+}
167+
168+func (t *TestDB) FindPipeMonitorHistory(monitorID string, from, to time.Time) ([]*db.PipeMonitorHistory, error) {
169+ return nil, nil
170+}
171+
99172 type TestSSHServer struct {
100- Cfg *shared.ConfigSite
101- DBPool *TestDB
102- Cancel context.CancelFunc
173+ Cfg *shared.ConfigSite
174+ DBPool *TestDB
175+ PipeHandler *CliHandler
176+ Cancel context.CancelFunc
103177 }
104178
105179 func NewTestSSHServer(t *testing.T) *TestSSHServer {
......@@ -178,9 +252,10 @@ func NewTestSSHServer(t *testing.T) *TestSSHServer {
178252 time.Sleep(100 * time.Millisecond)
179253
180254 return &TestSSHServer{
181- Cfg: cfg,
182- DBPool: dbpool,
183- Cancel: cancel,
255+ Cfg: cfg,
256+ DBPool: dbpool,
257+ PipeHandler: handler,
258+ Cancel: cancel,
184259 }
185260 }
186261
......@@ -1393,3 +1468,736 @@ func TestPubSub_MultipleSubscribers(t *testing.T) {
13931468 t.Errorf("subscriber 3 did not receive message, got: %q", string(received3[:n3]))
13941469 }
13951470 }
1471+
1472+// Monitor CLI Tests
1473+
1474+func TestMonitor_UnauthenticatedUserDenied(t *testing.T) {
1475+ server := NewTestSSHServer(t)
1476+ defer server.Shutdown()
1477+
1478+ user := GenerateUser("anonymous")
1479+
1480+ client, err := user.NewClient()
1481+ if err != nil {
1482+ t.Fatalf("failed to connect: %v", err)
1483+ }
1484+ defer func() { _ = client.Close() }()
1485+
1486+ output, err := user.RunCommand(client, "monitor my-service 1h")
1487+ if err != nil {
1488+ t.Logf("command error (expected): %v", err)
1489+ }
1490+
1491+ if !strings.Contains(output, "access denied") {
1492+ t.Errorf("expected 'access denied', got: %s", output)
1493+ }
1494+}
1495+
1496+func TestMonitor_CreateMonitor(t *testing.T) {
1497+ server := NewTestSSHServer(t)
1498+ defer server.Shutdown()
1499+
1500+ user := GenerateUser("alice")
1501+ RegisterUserWithServer(server, user)
1502+
1503+ client, err := user.NewClient()
1504+ if err != nil {
1505+ t.Fatalf("failed to connect: %v", err)
1506+ }
1507+ defer func() { _ = client.Close() }()
1508+
1509+ output, err := user.RunCommand(client, "monitor pico-uptime 24h")
1510+ if err != nil {
1511+ t.Logf("command completed: %v", err)
1512+ }
1513+
1514+ if strings.Contains(output, "access denied") {
1515+ t.Errorf("authenticated user should not get access denied, got: %s", output)
1516+ }
1517+
1518+ // Verify monitor was created in DB (topic is stored with user prefix)
1519+ monitor, err := server.DBPool.FindPipeMonitorByTopic("alice-id", "alice/pico-uptime")
1520+ if err != nil {
1521+ t.Fatalf("monitor should exist in DB: %v", err)
1522+ }
1523+
1524+ if monitor.WindowDur != 24*time.Hour {
1525+ t.Errorf("expected window duration 24h, got: %v", monitor.WindowDur)
1526+ }
1527+
1528+ if !strings.Contains(output, "alice/pico-uptime") || !strings.Contains(output, "24h") {
1529+ t.Errorf("output should confirm monitor creation, got: %s", output)
1530+ }
1531+}
1532+
1533+func TestMonitor_UpdateMonitor(t *testing.T) {
1534+ server := NewTestSSHServer(t)
1535+ defer server.Shutdown()
1536+
1537+ user := GenerateUser("alice")
1538+ RegisterUserWithServer(server, user)
1539+
1540+ client, err := user.NewClient()
1541+ if err != nil {
1542+ t.Fatalf("failed to connect: %v", err)
1543+ }
1544+ defer func() { _ = client.Close() }()
1545+
1546+ // Create initial monitor
1547+ _, err = user.RunCommand(client, "monitor my-cron 1h")
1548+ if err != nil {
1549+ t.Logf("create command completed: %v", err)
1550+ }
1551+
1552+ // Upsert with new duration
1553+ output, err := user.RunCommand(client, "monitor my-cron 6h")
1554+ if err != nil {
1555+ t.Logf("update command completed: %v", err)
1556+ }
1557+
1558+ // Verify monitor was updated (topic is stored with user prefix)
1559+ monitor, err := server.DBPool.FindPipeMonitorByTopic("alice-id", "alice/my-cron")
1560+ if err != nil {
1561+ t.Fatalf("monitor should exist in DB: %v", err)
1562+ }
1563+
1564+ if monitor.WindowDur != 6*time.Hour {
1565+ t.Errorf("expected window duration 6h after update, got: %v", monitor.WindowDur)
1566+ }
1567+
1568+ if !strings.Contains(output, "6h") {
1569+ t.Errorf("output should confirm updated duration, got: %s", output)
1570+ }
1571+}
1572+
1573+func TestMonitor_DeleteMonitor(t *testing.T) {
1574+ server := NewTestSSHServer(t)
1575+ defer server.Shutdown()
1576+
1577+ user := GenerateUser("alice")
1578+ RegisterUserWithServer(server, user)
1579+
1580+ client, err := user.NewClient()
1581+ if err != nil {
1582+ t.Fatalf("failed to connect: %v", err)
1583+ }
1584+ defer func() { _ = client.Close() }()
1585+
1586+ // Create monitor first
1587+ _, err = user.RunCommand(client, "monitor to-delete 1h")
1588+ if err != nil {
1589+ t.Logf("create command completed: %v", err)
1590+ }
1591+
1592+ // Verify it exists (topic is stored with user prefix)
1593+ _, err = server.DBPool.FindPipeMonitorByTopic("alice-id", "alice/to-delete")
1594+ if err != nil {
1595+ t.Fatalf("monitor should exist before deletion: %v", err)
1596+ }
1597+
1598+ // Delete it
1599+ output, err := user.RunCommand(client, "monitor to-delete -d")
1600+ if err != nil {
1601+ t.Logf("delete command completed: %v", err)
1602+ }
1603+
1604+ // Verify it's gone (topic is stored with user prefix)
1605+ _, err = server.DBPool.FindPipeMonitorByTopic("alice-id", "alice/to-delete")
1606+ if err == nil {
1607+ t.Errorf("monitor should be deleted from DB")
1608+ }
1609+
1610+ if !strings.Contains(output, "deleted") && !strings.Contains(output, "removed") {
1611+ t.Logf("output should confirm deletion, got: %s", output)
1612+ }
1613+}
1614+
1615+func TestMonitor_InvalidDuration(t *testing.T) {
1616+ server := NewTestSSHServer(t)
1617+ defer server.Shutdown()
1618+
1619+ user := GenerateUser("alice")
1620+ RegisterUserWithServer(server, user)
1621+
1622+ client, err := user.NewClient()
1623+ if err != nil {
1624+ t.Fatalf("failed to connect: %v", err)
1625+ }
1626+ defer func() { _ = client.Close() }()
1627+
1628+ output, err := user.RunCommand(client, "monitor my-service invaliduration")
1629+ if err != nil {
1630+ t.Logf("command error (expected): %v", err)
1631+ }
1632+
1633+ if !strings.Contains(output, "invalid") && !strings.Contains(output, "duration") && !strings.Contains(output, "error") {
1634+ t.Errorf("expected error about invalid duration, got: %s", output)
1635+ }
1636+}
1637+
1638+func TestMonitor_MissingTopic(t *testing.T) {
1639+ server := NewTestSSHServer(t)
1640+ defer server.Shutdown()
1641+
1642+ user := GenerateUser("alice")
1643+ RegisterUserWithServer(server, user)
1644+
1645+ client, err := user.NewClient()
1646+ if err != nil {
1647+ t.Fatalf("failed to connect: %v", err)
1648+ }
1649+ defer func() { _ = client.Close() }()
1650+
1651+ output, err := user.RunCommand(client, "monitor")
1652+ if err != nil {
1653+ t.Logf("command error (expected): %v", err)
1654+ }
1655+
1656+ // Should show usage or error about missing topic
1657+ if !strings.Contains(output, "Usage") && !strings.Contains(output, "topic") && !strings.Contains(output, "error") {
1658+ t.Errorf("expected usage info or error about missing topic, got: %s", output)
1659+ }
1660+}
1661+
1662+// Status CLI Tests
1663+
1664+func TestStatus_UnauthenticatedUserDenied(t *testing.T) {
1665+ server := NewTestSSHServer(t)
1666+ defer server.Shutdown()
1667+
1668+ user := GenerateUser("anonymous")
1669+
1670+ client, err := user.NewClient()
1671+ if err != nil {
1672+ t.Fatalf("failed to connect: %v", err)
1673+ }
1674+ defer func() { _ = client.Close() }()
1675+
1676+ output, err := user.RunCommand(client, "status")
1677+ if err != nil {
1678+ t.Logf("command error (expected): %v", err)
1679+ }
1680+
1681+ if !strings.Contains(output, "access denied") {
1682+ t.Errorf("expected 'access denied', got: %s", output)
1683+ }
1684+}
1685+
1686+func TestStatus_NoMonitors(t *testing.T) {
1687+ server := NewTestSSHServer(t)
1688+ defer server.Shutdown()
1689+
1690+ user := GenerateUser("alice")
1691+ RegisterUserWithServer(server, user)
1692+
1693+ client, err := user.NewClient()
1694+ if err != nil {
1695+ t.Fatalf("failed to connect: %v", err)
1696+ }
1697+ defer func() { _ = client.Close() }()
1698+
1699+ output, err := user.RunCommand(client, "status")
1700+ if err != nil {
1701+ t.Logf("command completed: %v", err)
1702+ }
1703+
1704+ if !strings.Contains(output, "no monitors") && !strings.Contains(output, "empty") {
1705+ t.Errorf("expected message about no monitors, got: %s", output)
1706+ }
1707+}
1708+
1709+func TestStatus_ShowsMonitorStatus(t *testing.T) {
1710+ server := NewTestSSHServer(t)
1711+ defer server.Shutdown()
1712+
1713+ user := GenerateUser("alice")
1714+ RegisterUserWithServer(server, user)
1715+
1716+ client, err := user.NewClient()
1717+ if err != nil {
1718+ t.Fatalf("failed to connect: %v", err)
1719+ }
1720+ defer func() { _ = client.Close() }()
1721+
1722+ // Create a monitor
1723+ _, err = user.RunCommand(client, "monitor web-check 1h")
1724+ if err != nil {
1725+ t.Logf("create monitor completed: %v", err)
1726+ }
1727+
1728+ // Check status
1729+ output, err := user.RunCommand(client, "status")
1730+ if err != nil {
1731+ t.Logf("status command completed: %v", err)
1732+ }
1733+
1734+ if !strings.Contains(output, "web-check") {
1735+ t.Errorf("status should list the monitor topic, got: %s", output)
1736+ }
1737+}
1738+
1739+func TestStatus_ShowsHealthyUnhealthy(t *testing.T) {
1740+ server := NewTestSSHServer(t)
1741+ defer server.Shutdown()
1742+
1743+ user := GenerateUser("alice")
1744+ RegisterUserWithServer(server, user)
1745+
1746+ // Create monitors directly in DB with different states
1747+ now := time.Now()
1748+ windowEnd := now.Add(1 * time.Hour)
1749+ recentPing := now.Add(-30 * time.Minute) // within window - healthy
1750+ oldPing := now.Add(-2 * time.Hour) // outside window - unhealthy
1751+
1752+ _ = server.DBPool.UpsertPipeMonitor("alice-id", "healthy-service", 1*time.Hour, &windowEnd)
1753+ _ = server.DBPool.UpdatePipeMonitorLastPing("alice-id", "healthy-service", &recentPing)
1754+
1755+ _ = server.DBPool.UpsertPipeMonitor("alice-id", "unhealthy-service", 1*time.Hour, &windowEnd)
1756+ _ = server.DBPool.UpdatePipeMonitorLastPing("alice-id", "unhealthy-service", &oldPing)
1757+
1758+ client, err := user.NewClient()
1759+ if err != nil {
1760+ t.Fatalf("failed to connect: %v", err)
1761+ }
1762+ defer func() { _ = client.Close() }()
1763+
1764+ output, err := user.RunCommand(client, "status")
1765+ if err != nil {
1766+ t.Logf("status command completed: %v", err)
1767+ }
1768+
1769+ if !strings.Contains(output, "healthy-service") {
1770+ t.Errorf("status should list healthy-service, got: %s", output)
1771+ }
1772+
1773+ if !strings.Contains(output, "unhealthy-service") {
1774+ t.Errorf("status should list unhealthy-service, got: %s", output)
1775+ }
1776+
1777+ // Should indicate different health states
1778+ if !strings.Contains(strings.ToLower(output), "healthy") && !strings.Contains(strings.ToLower(output), "ok") && !strings.Contains(output, "✓") {
1779+ t.Logf("status output should indicate health state: %s", output)
1780+ }
1781+}
1782+
1783+// RSS CLI Tests
1784+
1785+func TestRss_UnauthenticatedUserDenied(t *testing.T) {
1786+ server := NewTestSSHServer(t)
1787+ defer server.Shutdown()
1788+
1789+ user := GenerateUser("anonymous")
1790+
1791+ client, err := user.NewClient()
1792+ if err != nil {
1793+ t.Fatalf("failed to connect: %v", err)
1794+ }
1795+ defer func() { _ = client.Close() }()
1796+
1797+ output, err := user.RunCommand(client, "rss")
1798+ if err != nil {
1799+ t.Logf("command error (expected): %v", err)
1800+ }
1801+
1802+ if !strings.Contains(output, "access denied") {
1803+ t.Errorf("expected 'access denied', got: %s", output)
1804+ }
1805+}
1806+
1807+func TestRss_GeneratesValidRSS(t *testing.T) {
1808+ server := NewTestSSHServer(t)
1809+ defer server.Shutdown()
1810+
1811+ user := GenerateUser("alice")
1812+ RegisterUserWithServer(server, user)
1813+
1814+ // Create a monitor
1815+ now := time.Now()
1816+ windowEnd := now.Add(1 * time.Hour)
1817+ _ = server.DBPool.UpsertPipeMonitor("alice-id", "rss-test-service", 1*time.Hour, &windowEnd)
1818+
1819+ client, err := user.NewClient()
1820+ if err != nil {
1821+ t.Fatalf("failed to connect: %v", err)
1822+ }
1823+ defer func() { _ = client.Close() }()
1824+
1825+ output, err := user.RunCommand(client, "rss")
1826+ if err != nil {
1827+ t.Logf("rss command completed: %v", err)
1828+ }
1829+
1830+ // Should output valid RSS XML
1831+ if !strings.Contains(output, "<?xml") || !strings.Contains(output, "<rss") {
1832+ t.Errorf("expected RSS XML output, got: %s", output)
1833+ }
1834+
1835+ if !strings.Contains(output, "rss-test-service") {
1836+ t.Errorf("RSS should contain monitor topic, got: %s", output)
1837+ }
1838+}
1839+
1840+func TestRss_AlertsOnStaleMonitor(t *testing.T) {
1841+ server := NewTestSSHServer(t)
1842+ defer server.Shutdown()
1843+
1844+ user := GenerateUser("alice")
1845+ RegisterUserWithServer(server, user)
1846+
1847+ // Create a stale monitor (last ping outside window)
1848+ now := time.Now()
1849+ windowEnd := now.Add(-30 * time.Minute) // window already ended
1850+ oldPing := now.Add(-2 * time.Hour)
1851+
1852+ _ = server.DBPool.UpsertPipeMonitor("alice-id", "stale-service", 1*time.Hour, &windowEnd)
1853+ _ = server.DBPool.UpdatePipeMonitorLastPing("alice-id", "stale-service", &oldPing)
1854+
1855+ client, err := user.NewClient()
1856+ if err != nil {
1857+ t.Fatalf("failed to connect: %v", err)
1858+ }
1859+ defer func() { _ = client.Close() }()
1860+
1861+ output, err := user.RunCommand(client, "rss")
1862+ if err != nil {
1863+ t.Logf("rss command completed: %v", err)
1864+ }
1865+
1866+ // Should contain alert item for stale service
1867+ if !strings.Contains(output, "stale-service") {
1868+ t.Errorf("RSS should contain stale-service alert, got: %s", output)
1869+ }
1870+
1871+ // Should have item element for the alert
1872+ if !strings.Contains(output, "<item>") {
1873+ t.Errorf("RSS should contain item element for alert, got: %s", output)
1874+ }
1875+}
1876+
1877+// Pub integration with Monitor
1878+
1879+func TestPub_UpdatesMonitorLastPing(t *testing.T) {
1880+ server := NewTestSSHServer(t)
1881+ defer server.Shutdown()
1882+
1883+ user := GenerateUser("alice")
1884+ RegisterUserWithServer(server, user)
1885+
1886+ // Create a monitor first (topic is stored with user prefix)
1887+ now := time.Now()
1888+ windowEnd := now.Add(1 * time.Hour)
1889+ _ = server.DBPool.UpsertPipeMonitor("alice-id", "alice/ping-test", 1*time.Hour, &windowEnd)
1890+
1891+ subClient, err := user.NewClient()
1892+ if err != nil {
1893+ t.Fatalf("failed to connect subscriber: %v", err)
1894+ }
1895+ defer func() { _ = subClient.Close() }()
1896+
1897+ pubClient, err := user.NewClient()
1898+ if err != nil {
1899+ t.Fatalf("failed to connect publisher: %v", err)
1900+ }
1901+ defer func() { _ = pubClient.Close() }()
1902+
1903+ // Start subscriber
1904+ subSession, err := subClient.NewSession()
1905+ if err != nil {
1906+ t.Fatalf("failed to create sub session: %v", err)
1907+ }
1908+ defer func() { _ = subSession.Close() }()
1909+
1910+ if err := subSession.Start("sub ping-test -c"); err != nil {
1911+ t.Fatalf("failed to start sub: %v", err)
1912+ }
1913+
1914+ time.Sleep(100 * time.Millisecond)
1915+
1916+ // Publish to the monitored topic
1917+ _, err = user.RunCommandWithStdin(pubClient, "pub ping-test -c", "health check")
1918+ if err != nil {
1919+ t.Logf("pub command completed: %v", err)
1920+ }
1921+
1922+ // Verify last_ping was updated (topic is stored with user prefix)
1923+ monitor, err := server.DBPool.FindPipeMonitorByTopic("alice-id", "alice/ping-test")
1924+ if err != nil {
1925+ t.Fatalf("monitor should exist: %v", err)
1926+ }
1927+
1928+ if monitor.LastPing == nil {
1929+ t.Errorf("last_ping should be set after pub")
1930+ } else if time.Since(*monitor.LastPing) > 5*time.Second {
1931+ t.Errorf("last_ping should be recent, got: %v", monitor.LastPing)
1932+ }
1933+}
1934+
1935+// Tests for monitor status edge cases
1936+
1937+func TestStatus_PingAtExactWindowStart(t *testing.T) {
1938+ // Bug fix: Status() should use >= for windowStart comparison
1939+ // A ping exactly at windowStart should be healthy
1940+ now := time.Now().UTC()
1941+ windowEnd := now.Add(1 * time.Hour)
1942+ windowStart := windowEnd.Add(-1 * time.Hour) // equals now
1943+
1944+ monitor := &db.PipeMonitor{
1945+ LastPing: &windowStart, // ping exactly at window start
1946+ WindowEnd: &windowEnd,
1947+ WindowDur: 1 * time.Hour,
1948+ }
1949+
1950+ err := monitor.Status()
1951+ if err != nil {
1952+ t.Errorf("ping at exact window start should be healthy, got: %v", err)
1953+ }
1954+}
1955+
1956+func TestStatus_WindowExpired(t *testing.T) {
1957+ // Bug fix: Status() should check if current time is past windowEnd
1958+ now := time.Now().UTC()
1959+ windowEnd := now.Add(-1 * time.Minute) // window ended 1 minute ago
1960+ lastPing := now.Add(-30 * time.Second) // ping was 30 seconds ago
1961+
1962+ monitor := &db.PipeMonitor{
1963+ LastPing: &lastPing,
1964+ WindowEnd: &windowEnd,
1965+ WindowDur: 1 * time.Hour,
1966+ }
1967+
1968+ err := monitor.Status()
1969+ if err == nil {
1970+ t.Error("expired window should be unhealthy")
1971+ }
1972+ if !strings.Contains(err.Error(), "window expired") {
1973+ t.Errorf("error should mention window expired, got: %v", err)
1974+ }
1975+}
1976+
1977+func TestStatus_PingResetsWindow(t *testing.T) {
1978+ // Bug fix: Every ping should reset window to now + duration
1979+ server := NewTestSSHServer(t)
1980+ defer server.Shutdown()
1981+
1982+ user := GenerateUser("alice")
1983+ RegisterUserWithServer(server, user)
1984+
1985+ // Create a monitor with an expired window
1986+ expiredWindowEnd := time.Now().UTC().Add(-10 * time.Minute)
1987+ _ = server.DBPool.UpsertPipeMonitor("alice-id", "alice/reset-test", 5*time.Minute, &expiredWindowEnd)
1988+
1989+ client, err := user.NewClient()
1990+ if err != nil {
1991+ t.Fatalf("failed to connect: %v", err)
1992+ }
1993+ defer func() { _ = client.Close() }()
1994+
1995+ // Start a subscriber first so pub doesn't block
1996+ subClient, err := user.NewClient()
1997+ if err != nil {
1998+ t.Fatalf("failed to connect subscriber: %v", err)
1999+ }
2000+ defer func() { _ = subClient.Close() }()
2001+
2002+ subSession, err := subClient.NewSession()
2003+ if err != nil {
2004+ t.Fatalf("failed to create sub session: %v", err)
2005+ }
2006+ defer func() { _ = subSession.Close() }()
2007+
2008+ if err := subSession.Start("sub reset-test -c"); err != nil {
2009+ t.Fatalf("failed to start sub: %v", err)
2010+ }
2011+
2012+ time.Sleep(100 * time.Millisecond)
2013+
2014+ // Pub to trigger monitor update
2015+ _, err = user.RunCommandWithStdin(client, "pub reset-test -c", "ping")
2016+ if err != nil {
2017+ t.Logf("pub command completed: %v", err)
2018+ }
2019+
2020+ // Check that window was reset
2021+ monitor, err := server.DBPool.FindPipeMonitorByTopic("alice-id", "alice/reset-test")
2022+ if err != nil {
2023+ t.Fatalf("monitor should exist: %v", err)
2024+ }
2025+
2026+ if monitor.WindowEnd == nil {
2027+ t.Fatal("window_end should be set")
2028+ }
2029+
2030+ // Window end should now be in the future
2031+ if !monitor.WindowEnd.After(time.Now().UTC()) {
2032+ t.Errorf("window_end should be in the future after ping, got: %v", monitor.WindowEnd)
2033+ }
2034+}
2035+
2036+func TestStatus_HealthyImmediatelyAfterPing(t *testing.T) {
2037+ // Bug fix: After a ping, status should immediately show healthy
2038+ server := NewTestSSHServer(t)
2039+ defer server.Shutdown()
2040+
2041+ user := GenerateUser("alice")
2042+ RegisterUserWithServer(server, user)
2043+
2044+ client, err := user.NewClient()
2045+ if err != nil {
2046+ t.Fatalf("failed to connect: %v", err)
2047+ }
2048+ defer func() { _ = client.Close() }()
2049+
2050+ // Create monitor
2051+ _, err = user.RunCommand(client, "monitor health-test 5m")
2052+ if err != nil {
2053+ t.Fatalf("failed to create monitor: %v", err)
2054+ }
2055+
2056+ // Start subscriber
2057+ subClient, err := user.NewClient()
2058+ if err != nil {
2059+ t.Fatalf("failed to connect subscriber: %v", err)
2060+ }
2061+ defer func() { _ = subClient.Close() }()
2062+
2063+ subSession, err := subClient.NewSession()
2064+ if err != nil {
2065+ t.Fatalf("failed to create sub session: %v", err)
2066+ }
2067+ defer func() { _ = subSession.Close() }()
2068+
2069+ if err := subSession.Start("sub health-test -c"); err != nil {
2070+ t.Fatalf("failed to start sub: %v", err)
2071+ }
2072+
2073+ time.Sleep(100 * time.Millisecond)
2074+
2075+ // Pub to trigger ping
2076+ pubClient, err := user.NewClient()
2077+ if err != nil {
2078+ t.Fatalf("failed to connect publisher: %v", err)
2079+ }
2080+ defer func() { _ = pubClient.Close() }()
2081+
2082+ _, err = user.RunCommandWithStdin(pubClient, "pub health-test -c", "ping")
2083+ if err != nil {
2084+ t.Logf("pub completed: %v", err)
2085+ }
2086+
2087+ // Immediately check status
2088+ statusClient, err := user.NewClient()
2089+ if err != nil {
2090+ t.Fatalf("failed to connect for status: %v", err)
2091+ }
2092+ defer func() { _ = statusClient.Close() }()
2093+
2094+ output, err := user.RunCommand(statusClient, "status")
2095+ if err != nil {
2096+ t.Logf("status completed: %v", err)
2097+ }
2098+
2099+ if strings.Contains(output, "unhealthy") {
2100+ t.Errorf("status should be healthy immediately after ping, got: %s", output)
2101+ }
2102+ if !strings.Contains(output, "healthy") {
2103+ t.Errorf("status should show healthy, got: %s", output)
2104+ }
2105+}
2106+
2107+// TestMonitor_FixedWindowNonSliding verifies that pings within the same window
2108+// do not slide the window forward. This is a regression test for a bug where
2109+// each ping reset window_end to now+dur, creating a sliding window that never fails.
2110+//
2111+// Expected behavior:
2112+// - last_ping: always updated to show most recent activity (user visibility).
2113+// - window_end: only advances when current time exceeds it (health scheduling).
2114+func TestMonitor_FixedWindowNonSliding(t *testing.T) {
2115+ server := NewTestSSHServer(t)
2116+ defer server.Shutdown()
2117+
2118+ user := GenerateUser("alice")
2119+ RegisterUserWithServer(server, user)
2120+
2121+ client, err := user.NewClient()
2122+ if err != nil {
2123+ t.Fatalf("failed to connect: %v", err)
2124+ }
2125+ defer func() { _ = client.Close() }()
2126+
2127+ // Create a monitor with 1 hour window
2128+ _, err = user.RunCommand(client, "monitor fixed-window-test 1h")
2129+ if err != nil {
2130+ t.Logf("create command completed: %v", err)
2131+ }
2132+
2133+ // Get the initial window_end
2134+ monitor, err := server.DBPool.FindPipeMonitorByTopic("alice-id", "alice/fixed-window-test")
2135+ if err != nil {
2136+ t.Fatalf("monitor should exist: %v", err)
2137+ }
2138+ initialWindowEnd := *monitor.WindowEnd
2139+
2140+ // Simulate a ping by calling updateMonitor directly
2141+ handler := server.PipeHandler
2142+
2143+ // Create a mock CliCmd
2144+ mockUser := &db.User{ID: "alice-id", Name: "alice"}
2145+ cmd := &CliCmd{
2146+ userName: "alice",
2147+ user: mockUser,
2148+ }
2149+
2150+ // First ping - should record last_ping but NOT change window_end
2151+ handler.updateMonitor(cmd, "alice/fixed-window-test")
2152+
2153+ monitor, err = server.DBPool.FindPipeMonitorByTopic("alice-id", "alice/fixed-window-test")
2154+ if err != nil {
2155+ t.Fatalf("monitor should exist after first ping: %v", err)
2156+ }
2157+
2158+ if monitor.LastPing == nil {
2159+ t.Fatalf("last_ping should be set after first ping")
2160+ }
2161+ firstPingTime := *monitor.LastPing
2162+ windowEndAfterFirstPing := *monitor.WindowEnd
2163+
2164+ // BUG CHECK: With the bug, window_end would have slid forward to now+1h
2165+ // With the fix, window_end should remain at the original scheduled time
2166+ if !windowEndAfterFirstPing.Equal(initialWindowEnd) {
2167+ t.Errorf("BUG DETECTED: window_end should NOT change after first ping within window\n"+
2168+ "initial window_end: %v\n"+
2169+ "window_end after ping: %v\n"+
2170+ "Window slid forward by: %v",
2171+ initialWindowEnd.Format(time.RFC3339),
2172+ windowEndAfterFirstPing.Format(time.RFC3339),
2173+ windowEndAfterFirstPing.Sub(initialWindowEnd))
2174+ }
2175+
2176+ // Second ping - last_ping SHOULD be updated (for user visibility)
2177+ // but window_end should NOT change
2178+ time.Sleep(10 * time.Millisecond) // Small delay to get different timestamp
2179+ handler.updateMonitor(cmd, "alice/fixed-window-test")
2180+
2181+ monitor, err = server.DBPool.FindPipeMonitorByTopic("alice-id", "alice/fixed-window-test")
2182+ if err != nil {
2183+ t.Fatalf("monitor should exist after second ping: %v", err)
2184+ }
2185+
2186+ // last_ping SHOULD be updated to show most recent activity
2187+ if monitor.LastPing.Equal(firstPingTime) {
2188+ t.Errorf("last_ping SHOULD be updated for user visibility\n"+
2189+ "first ping time: %v\n"+
2190+ "last_ping after second call: %v",
2191+ firstPingTime.Format(time.RFC3339Nano),
2192+ monitor.LastPing.Format(time.RFC3339Nano))
2193+ }
2194+
2195+ // But window_end should still be the original value (not sliding)
2196+ if !monitor.WindowEnd.Equal(initialWindowEnd) {
2197+ t.Errorf("BUG DETECTED: window_end should remain at original value\n"+
2198+ "initial: %v\n"+
2199+ "current: %v",
2200+ initialWindowEnd.Format(time.RFC3339),
2201+ monitor.WindowEnd.Format(time.RFC3339))
2202+ }
2203+}
M pkg/db/db.go
+141, -0
......@@ -5,6 +5,7 @@ import (
55 "database/sql/driver"
66 "encoding/json"
77 "errors"
8+ "fmt"
89 "regexp"
910 "time"
1011 )
......@@ -378,6 +379,137 @@ type TunsEventLog struct {
378379 CreatedAt *time.Time `json:"created_at" db:"created_at"`
379380 }
380381
382+type PipeMonitor struct {
383+ ID string `json:"id" db:"id"`
384+ UserId string `json:"user_id" db:"user_id"`
385+ Topic string `json:"topic" db:"topic"`
386+ WindowDur time.Duration `json:"window_dur" db:"window_dur"`
387+ WindowEnd *time.Time `json:"window_end" db:"window_end"`
388+ LastPing *time.Time `json:"last_ping" db:"last_ping"`
389+ CreatedAt *time.Time `json:"created_at" db:"created_at"`
390+ UpdatedAt *time.Time `json:"updated_at" db:"updated_at"`
391+}
392+
393+type PipeMonitorHistory struct {
394+ ID string `json:"id" db:"id"`
395+ MonitorID string `json:"monitor_id" db:"monitor_id"`
396+ WindowDur time.Duration `json:"window_dur" db:"window_dur"`
397+ WindowEnd *time.Time `json:"window_end" db:"window_end"`
398+ LastPing *time.Time `json:"last_ping" db:"last_ping"`
399+ CreatedAt *time.Time `json:"created_at" db:"created_at"`
400+ UpdatedAt *time.Time `json:"updated_at" db:"updated_at"`
401+}
402+
403+type UptimeResult struct {
404+ TotalDuration time.Duration
405+ UptimeDuration time.Duration
406+ UptimePercent float64
407+}
408+
409+func ComputeUptime(history []*PipeMonitorHistory, from, to time.Time) UptimeResult {
410+ totalDuration := to.Sub(from)
411+ if totalDuration <= 0 {
412+ return UptimeResult{}
413+ }
414+
415+ if len(history) == 0 {
416+ return UptimeResult{TotalDuration: totalDuration}
417+ }
418+
419+ type interval struct {
420+ start, end time.Time
421+ }
422+
423+ var intervals []interval
424+ for _, h := range history {
425+ if h.WindowEnd == nil {
426+ continue
427+ }
428+ windowStart := h.WindowEnd.Add(-h.WindowDur)
429+ windowEnd := *h.WindowEnd
430+
431+ if windowStart.Before(from) {
432+ windowStart = from
433+ }
434+ if windowEnd.After(to) {
435+ windowEnd = to
436+ }
437+
438+ if windowStart.Before(windowEnd) {
439+ intervals = append(intervals, interval{start: windowStart, end: windowEnd})
440+ }
441+ }
442+
443+ if len(intervals) == 0 {
444+ return UptimeResult{TotalDuration: totalDuration}
445+ }
446+
447+ // Sort by start time
448+ for i := range intervals {
449+ for j := i + 1; j < len(intervals); j++ {
450+ if intervals[j].start.Before(intervals[i].start) {
451+ intervals[i], intervals[j] = intervals[j], intervals[i]
452+ }
453+ }
454+ }
455+
456+ // Merge overlapping intervals
457+ merged := []interval{intervals[0]}
458+ for _, curr := range intervals[1:] {
459+ last := &merged[len(merged)-1]
460+ if !curr.start.After(last.end) {
461+ if curr.end.After(last.end) {
462+ last.end = curr.end
463+ }
464+ } else {
465+ merged = append(merged, curr)
466+ }
467+ }
468+
469+ var uptimeDuration time.Duration
470+ for _, iv := range merged {
471+ uptimeDuration += iv.end.Sub(iv.start)
472+ }
473+
474+ uptimePercent := float64(uptimeDuration) / float64(totalDuration) * 100
475+
476+ return UptimeResult{
477+ TotalDuration: totalDuration,
478+ UptimeDuration: uptimeDuration,
479+ UptimePercent: uptimePercent,
480+ }
481+}
482+
483+func (m *PipeMonitor) Status() error {
484+ if m.LastPing == nil {
485+ return fmt.Errorf("no ping received yet")
486+ }
487+ if m.WindowEnd == nil {
488+ return fmt.Errorf("window end not set")
489+ }
490+ now := time.Now().UTC()
491+ if now.After(*m.WindowEnd) {
492+ return fmt.Errorf(
493+ "window expired at %s",
494+ m.WindowEnd.UTC().Format("2006-01-02 15:04:05Z"),
495+ )
496+ }
497+ windowStart := m.WindowEnd.Add(-m.WindowDur)
498+ lastPingAfterStart := !m.LastPing.Before(windowStart)
499+ if !lastPingAfterStart {
500+ return fmt.Errorf(
501+ "last ping before window start: %s",
502+ windowStart.UTC().Format("2006-01-02 15:04:05Z"),
503+ )
504+ }
505+ return nil
506+}
507+
508+func (m *PipeMonitor) GetNextWindow() *time.Time {
509+ win := m.WindowEnd.Add(m.WindowDur)
510+ return &win
511+}
512+
381513 var NameValidator = regexp.MustCompile("^[a-zA-Z0-9]{1,50}$")
382514 var DenyList = []string{
383515 "admin",
......@@ -468,5 +600,14 @@ type DB interface {
468600 FindPubkeysInAccessLogs(userID string) ([]string, error)
469601 FindAccessLogsByPubkey(pubkey string, fromDate *time.Time) ([]*AccessLog, error)
470602
603+ UpsertPipeMonitor(userID, topic string, dur time.Duration, winEnd *time.Time) error
604+ UpdatePipeMonitorLastPing(userID, topic string, lastPing *time.Time) error
605+ RemovePipeMonitor(userID, topic string) error
606+ FindPipeMonitorByTopic(userID, topic string) (*PipeMonitor, error)
607+ FindPipeMonitorsByUser(userID string) ([]*PipeMonitor, error)
608+
609+ InsertPipeMonitorHistory(monitorID string, windowDur time.Duration, windowEnd, lastPing *time.Time) error
610+ FindPipeMonitorHistory(monitorID string, from, to time.Time) ([]*PipeMonitorHistory, error)
611+
471612 Close() error
472613 }
M pkg/db/postgres/storage.go
+73, -0
......@@ -1519,3 +1519,76 @@ func (me *PsqlDB) InsertAccessLog(log *db.AccessLog) error {
15191519 )
15201520 return err
15211521 }
1522+
1523+func (me *PsqlDB) UpsertPipeMonitor(userID, topic string, dur time.Duration, winEnd *time.Time) error {
1524+ durStr := fmt.Sprintf("%d seconds", int64(dur.Seconds()))
1525+ _, err := me.Db.Exec(
1526+ `INSERT INTO pipe_monitors (user_id, topic, window_dur, window_end)
1527+ VALUES ($1, $2, $3::interval, $4)
1528+ ON CONFLICT (user_id, topic) DO UPDATE SET window_dur = $3::interval, window_end = $4, updated_at = NOW();`,
1529+ userID,
1530+ topic,
1531+ durStr,
1532+ winEnd,
1533+ )
1534+ return err
1535+}
1536+
1537+func (me *PsqlDB) UpdatePipeMonitorLastPing(userID, topic string, lastPing *time.Time) error {
1538+ _, err := me.Db.Exec(
1539+ `UPDATE pipe_monitors SET last_ping = $3, updated_at = NOW() WHERE user_id = $1 AND topic = $2;`,
1540+ userID,
1541+ topic,
1542+ lastPing,
1543+ )
1544+ return err
1545+}
1546+
1547+func (me *PsqlDB) RemovePipeMonitor(userID, topic string) error {
1548+ _, err := me.Db.Exec(
1549+ `DELETE FROM pipe_monitors WHERE user_id = $1 AND topic = $2;`,
1550+ userID,
1551+ topic,
1552+ )
1553+ return err
1554+}
1555+
1556+func (me *PsqlDB) FindPipeMonitorByTopic(userID, topic string) (*db.PipeMonitor, error) {
1557+ monitor := &db.PipeMonitor{}
1558+ 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)
1559+ if err != nil {
1560+ return nil, err
1561+ }
1562+ return monitor, nil
1563+}
1564+
1565+func (me *PsqlDB) FindPipeMonitorsByUser(userID string) ([]*db.PipeMonitor, error) {
1566+ var monitors []*db.PipeMonitor
1567+ 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)
1568+ if err != nil {
1569+ return nil, err
1570+ }
1571+ return monitors, nil
1572+}
1573+
1574+func (me *PsqlDB) InsertPipeMonitorHistory(monitorID string, windowDur time.Duration, windowEnd, lastPing *time.Time) error {
1575+ durStr := fmt.Sprintf("%d seconds", int64(windowDur.Seconds()))
1576+ _, err := me.Db.Exec(
1577+ `INSERT INTO pipe_monitors_history (monitor_id, window_dur, window_end, last_ping) VALUES ($1, $2::interval, $3, $4)`,
1578+ monitorID, durStr, windowEnd, lastPing,
1579+ )
1580+ return err
1581+}
1582+
1583+func (me *PsqlDB) FindPipeMonitorHistory(monitorID string, from, to time.Time) ([]*db.PipeMonitorHistory, error) {
1584+ var history []*db.PipeMonitorHistory
1585+ err := me.Db.Select(
1586+ &history,
1587+ `SELECT id, monitor_id, (EXTRACT(EPOCH FROM window_dur) * 1000000000)::bigint as window_dur, window_end, last_ping, created_at, updated_at FROM pipe_monitors_history WHERE monitor_id = $1 AND last_ping <= $2 AND window_end >= $3 ORDER BY last_ping ASC`,
1588+ monitorID, to, from,
1589+ )
1590+ if err != nil {
1591+ return nil, err
1592+ }
1593+ return history, nil
1594+}
M pkg/db/postgres/storage_test.go
+157, -1
......@@ -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,159 @@ 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+}
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+}
M pkg/db/stub/stub.go
+28, -0
......@@ -243,3 +243,31 @@ 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+}
262+
263+func (me *StubDB) FindPipeMonitorsByUser(userID string) ([]*db.PipeMonitor, error) {
264+ return nil, errNotImpl
265+}
266+
267+func (me *StubDB) InsertPipeMonitorHistory(monitorID string, windowDur time.Duration, windowEnd, lastPing *time.Time) error {
268+ return errNotImpl
269+}
270+
271+func (me *StubDB) FindPipeMonitorHistory(monitorID string, from, to time.Time) ([]*db.PipeMonitorHistory, error) {
272+ return nil, errNotImpl
273+}
A sql/migrations/20251226_add_pipe_monitoring.sql
+36, -0
......@@ -0,0 +1,36 @@
1+CREATE TABLE IF NOT EXISTS pipe_monitors (
2+ id uuid NOT NULL DEFAULT uuid_generate_v4(),
3+ user_id uuid NOT NULL,
4+ topic text NOT NULL,
5+ window_dur interval NOT NULL,
6+ window_end timestamp without time zone NOT NULL DEFAULT NOW(),
7+ last_ping timestamp,
8+ created_at timestamp without time zone NOT NULL DEFAULT NOW(),
9+ updated_at timestamp without time zone NOT NULL DEFAULT NOW(),
10+ CONSTRAINT pipe_monitors_unique_topic UNIQUE (user_id, topic),
11+ CONSTRAINT pipe_monitoring_pkey PRIMARY KEY (id),
12+ CONSTRAINT fk_pipe_monitoring_app_users
13+ FOREIGN KEY(user_id)
14+ REFERENCES app_users(id)
15+ ON DELETE CASCADE
16+ ON UPDATE CASCADE
17+);
18+
19+CREATE TABLE IF NOT EXISTS pipe_monitors_history (
20+ id uuid NOT NULL DEFAULT uuid_generate_v4(),
21+ monitor_id uuid NOT NULL,
22+ window_dur interval NOT NULL,
23+ window_end timestamp without time zone NOT NULL DEFAULT NOW(),
24+ last_ping timestamp,
25+ created_at timestamp without time zone NOT NULL DEFAULT NOW(),
26+ updated_at timestamp without time zone NOT NULL DEFAULT NOW(),
27+ CONSTRAINT pipe_monitor_history_pkey PRIMARY KEY (id),
28+ CONSTRAINT fk_pipe_monitor_history_pipe_monitors
29+ FOREIGN KEY(monitor_id)
30+ REFERENCES pipe_monitors(id)
31+ ON DELETE CASCADE
32+ ON UPDATE CASCADE
33+);
34+
35+CREATE INDEX IF NOT EXISTS idx_pipe_mon_hist_monitor_last_ping ON pipe_monitors_history (monitor_id, last_ping);
36+CREATE INDEX IF NOT EXISTS idx_pipe_mon_hist_monitor_window_end ON pipe_monitors_history (monitor_id, window_end);