Antonio Mika
·
2026-10-11
1package pipe
2
3import (
4 "bytes"
5 "context"
6 "database/sql"
7 "errors"
8 "flag"
9 "fmt"
10 "io"
11 "log/slog"
12 "slices"
13 "strings"
14 "sync/atomic"
15 "text/tabwriter"
16 "time"
17
18 "github.com/antoniomika/syncmap"
19 "github.com/google/uuid"
20 "github.com/gorilla/feeds"
21 "github.com/picosh/pico/pkg/db"
22 "github.com/picosh/pico/pkg/pssh"
23 psub "github.com/picosh/pico/pkg/pubsub"
24 gossh "golang.org/x/crypto/ssh"
25)
26
27func Middleware(handler *CliHandler) pssh.SSHServerMiddleware {
28 return func(next pssh.SSHServerHandler) pssh.SSHServerHandler {
29 return func(sesh *pssh.SSHServerConnSession) error {
30 ctx := sesh.Context()
31 logger := pssh.GetLogger(sesh)
32 user := pssh.GetUser(sesh)
33
34 args := sesh.Command()
35 if len(args) == 0 {
36 help(handler.Cfg, sesh)
37 return next(sesh)
38 }
39
40 userName := "public"
41 userNameAddition := ""
42 uuidStr := uuid.NewString()
43 isAdmin := false
44 if user != nil {
45 isAdmin = db.HasFeature(handler.DBPool, user.ID, "admin")
46 if isAdmin && strings.HasPrefix(sesh.User(), "admin__") {
47 uuidStr = fmt.Sprintf("admin-%s", uuidStr)
48 }
49
50 userName = user.Name
51 if user.PublicKey != nil && user.PublicKey.Name != "" {
52 addition := user.PublicKey.Name
53 identity := sesh.Permissions().Extensions["identity"]
54 if identity != "" && identity != "pubkey" {
55 addition = identity
56 }
57 userNameAddition = fmt.Sprintf("-%s", addition)
58 }
59 }
60
61 pipeCtx, cancel := context.WithCancel(ctx)
62
63 cliCmd := &CliCmd{
64 sesh: sesh,
65 args: args,
66 userName: userName,
67 isAdmin: isAdmin,
68 pipeCtx: pipeCtx,
69 cancel: cancel,
70 user: user,
71 }
72
73 cmd := strings.TrimSpace(args[0])
74 switch cmd {
75 case "help":
76 help(handler.Cfg, sesh)
77 return next(sesh)
78 case "ls":
79 err := handler.ls(cliCmd)
80 if err != nil {
81 logger.Error("ls cmd", "err", err)
82 sesh.Fatal(err)
83 }
84 return next(sesh)
85 case "monitor":
86 err := handler.monitor(cliCmd, user)
87 if err != nil {
88 logger.Error("monitor cmd", "err", err)
89 sesh.Fatal(err)
90 }
91 return next(sesh)
92 case "status":
93 err := handler.status(cliCmd, user)
94 if err != nil {
95 logger.Error("status cmd", "err", err)
96 sesh.Fatal(err)
97 }
98 return next(sesh)
99 case "rss":
100 rss, err := MonitorRss(handler.DBPool, user, handler.Cfg.Domain)
101 _, _ = fmt.Fprintln(sesh, rss)
102 if err != nil {
103 logger.Error("rss cmd", "err", err)
104 sesh.Fatal(err)
105 }
106 return next(sesh)
107 }
108
109 topic := ""
110 cmdArgs := args[1:]
111 if len(args) > 1 && !strings.HasPrefix(args[1], "-") {
112 topic = strings.TrimSpace(args[1])
113 cmdArgs = args[2:]
114 }
115 // sub commands after this line expect clipped args
116 cliCmd.args = cmdArgs
117
118 logger.Info(
119 "pubsub middleware detected command",
120 "args", args,
121 "cmd", cmd,
122 "topic", topic,
123 "cmdArgs", cmdArgs,
124 )
125
126 clientID := fmt.Sprintf(
127 "%s (%s%s@%s)",
128 uuidStr,
129 userName,
130 userNameAddition,
131 sesh.RemoteAddr().String(),
132 )
133
134 go func() {
135 defer cancel()
136
137 ticker := time.NewTicker(5 * time.Second)
138 defer ticker.Stop()
139
140 for {
141 select {
142 case <-pipeCtx.Done():
143 return
144 case <-ticker.C:
145 _, err := sesh.SendRequest("ping@pico.sh", false, nil)
146 if err != nil {
147 logger.Error("error sending ping", "err", err)
148 return
149 }
150 }
151 }
152 }()
153
154 switch cmd {
155 case "pub":
156 err := handler.pub(cliCmd, topic, clientID)
157 if err != nil {
158 logger.Error("pub cmd", "err", err)
159 sesh.Fatal(err)
160 }
161 case "sub":
162 err := handler.sub(cliCmd, topic, clientID)
163 if err != nil {
164 logger.Error("sub cmd", "err", err)
165 sesh.Fatal(err)
166 }
167 case "pipe":
168 err := handler.pipe(cliCmd, topic, clientID)
169 if err != nil {
170 logger.Error("pipe cmd", "err", err)
171 sesh.Fatal(err)
172 }
173 case "uptime":
174 err := handler.uptime(cliCmd, topic, user)
175 if err != nil {
176 logger.Error("uptime cmd", "err", err)
177 sesh.Fatal(err)
178 }
179 }
180
181 return next(sesh)
182 }
183 }
184}
185
186type CliHandler struct {
187 DBPool DB
188 Logger *slog.Logger
189 PubSub psub.PubSub
190 Cfg *Config
191 Waiters *syncmap.Map[string, []string]
192 Access *syncmap.Map[string, []string]
193}
194
195func (h *CliHandler) GetLogger(s *pssh.SSHServerConnSession) *slog.Logger {
196 return h.Logger
197}
198
199type CliCmd struct {
200 sesh *pssh.SSHServerConnSession
201 args []string
202 userName string
203 isAdmin bool
204 pipeCtx context.Context
205 cancel context.CancelFunc
206 user *db.User
207}
208
209func help(cfg *Config, sesh *pssh.SSHServerConnSession) {
210 data := fmt.Sprintf(`Command: ssh %s <command> [args...]
211
212The simplest authenticated pubsub system. Send messages through
213user-defined topics. Topics are private to the authenticated
214ssh user. The default pubsub model is multicast with bidirectional
215blocking, meaning a publisher ("pub") will send its message to all
216subscribers ("sub"). Further, both "pub" and "sub" will wait for
217at least one event to be sent or received. Pipe ("pipe") allows
218for bidirectional messages to be sent between any clients connected
219to a pipe.
220
221Commands:
222 help Show this help message
223 ls List active pubsub channels
224 pub <topic> [flags] Publish messages to a topic
225 sub <topic> [flags] Subscribe to messages from a topic
226 pipe <topic> [flags] Bidirectional messaging between clients
227
228Monitoring commands:
229 monitor <topic> <duration> Create/update a health monitor for a topic
230 monitor <topic> -d Delete a monitor
231 status Show health status of all monitors
232 uptime Show uptime for a topic
233 rss Get RSS feed of monitor alerts
234
235Use "ssh %s <command> -h" for help on a specific command.
236`, toSshCmd(cfg), toSshCmd(cfg))
237
238 data = strings.ReplaceAll(data, "\n", "\r\n")
239 _, _ = fmt.Fprintln(sesh, data)
240}
241
242func (handler *CliHandler) ls(cmd *CliCmd) error {
243 if cmd.userName == "public" {
244 err := fmt.Errorf("access denied")
245 return err
246 }
247
248 topicFilter := fmt.Sprintf("%s/", cmd.userName)
249 if cmd.isAdmin {
250 topicFilter = ""
251 if len(cmd.args) > 1 {
252 topicFilter = cmd.args[1]
253 }
254 }
255
256 var channels []*psub.Channel
257 waitingChannels := map[string][]string{}
258
259 for topic, channel := range handler.PubSub.GetChannels() {
260 if strings.HasPrefix(topic, topicFilter) {
261 channels = append(channels, channel)
262 }
263 }
264
265 for channel, clients := range handler.Waiters.Range {
266 if strings.HasPrefix(channel, topicFilter) {
267 waitingChannels[channel] = clients
268 }
269 }
270
271 if len(channels) == 0 && len(waitingChannels) == 0 {
272 _, _ = fmt.Fprintln(cmd.sesh, "no pubsub channels found")
273 } else {
274 var outputData string
275 if len(channels) > 0 || len(waitingChannels) > 0 {
276 outputData += "Channel Information\r\n"
277 for _, channel := range channels {
278 extraData := ""
279
280 if accessList, ok := handler.Access.Load(channel.Topic); ok && len(accessList) > 0 {
281 extraData += fmt.Sprintf(" (Access List: %s)", strings.Join(accessList, ", "))
282 }
283
284 outputData += fmt.Sprintf("- %s:%s\r\n", channel.Topic, extraData)
285
286 var pubs []*psub.Client
287 var subs []*psub.Client
288 var pipes []*psub.Client
289
290 for _, client := range channel.GetClients() {
291 switch client.Direction {
292 case psub.ChannelDirectionInput:
293 pubs = append(pubs, client)
294 case psub.ChannelDirectionOutput:
295 subs = append(subs, client)
296 case psub.ChannelDirectionInputOutput:
297 pipes = append(pipes, client)
298 }
299 }
300 outputData += clientInfo(pubs, cmd.isAdmin, "Pubs")
301 outputData += clientInfo(subs, cmd.isAdmin, "Subs")
302 outputData += clientInfo(pipes, cmd.isAdmin, "Pipes")
303 }
304
305 for waitingChannel, channelPubs := range waitingChannels {
306 extraData := ""
307
308 if accessList, ok := handler.Access.Load(waitingChannel); ok && len(accessList) > 0 {
309 extraData += fmt.Sprintf(" (Access List: %s)", strings.Join(accessList, ", "))
310 }
311
312 outputData += fmt.Sprintf("- %s:%s\r\n", waitingChannel, extraData)
313 outputData += fmt.Sprintf(" %s:\r\n", "Waiting Pubs")
314 for _, client := range channelPubs {
315 if strings.HasPrefix(client, "admin-") && !cmd.isAdmin {
316 continue
317 }
318 outputData += fmt.Sprintf(" - %s\r\n", client)
319 }
320 }
321 }
322
323 _, _ = cmd.sesh.Write([]byte(outputData))
324 }
325
326 return nil
327}
328
329func (handler *CliHandler) monitor(cmd *CliCmd, user *db.User) error {
330 if user == nil {
331 return fmt.Errorf("access denied")
332 }
333
334 args := cmd.sesh.Command()
335 topic := ""
336 cmdArgs := args[1:]
337 if len(args) > 1 && !strings.HasPrefix(args[1], "-") {
338 topic = strings.TrimSpace(args[1])
339 cmdArgs = args[2:]
340 }
341
342 monitorCmd := flagSet("monitor", cmd.sesh)
343 del := monitorCmd.Bool("d", false, "Delete the monitor")
344
345 if !flagCheck(monitorCmd, topic, cmdArgs) {
346 return nil
347 }
348
349 if topic == "" {
350 _, _ = fmt.Fprintln(cmd.sesh, "Usage: monitor <topic> <duration>")
351 _, _ = fmt.Fprintln(cmd.sesh, " monitor <topic> -d")
352 return fmt.Errorf("topic is required")
353 }
354
355 // Resolve to fully qualified topic name
356 result := resolveTopic(TopicResolveInput{
357 UserName: cmd.userName,
358 Topic: topic,
359 IsAdmin: cmd.isAdmin,
360 IsPublic: false,
361 })
362 resolvedTopic := result.Name
363
364 if *del {
365 handler.Logger.Info("removing pipe monitor", "topic", resolvedTopic)
366 err := handler.DBPool.RemovePipeMonitor(user.ID, resolvedTopic)
367 if err != nil {
368 return fmt.Errorf("failed to delete monitor: %w", err)
369 }
370 _, _ = fmt.Fprintf(cmd.sesh, "monitor deleted: %s\r\n", resolvedTopic)
371 return nil
372 }
373
374 // Create/update monitor - need duration argument
375 durStr := ""
376 if monitorCmd.NArg() > 0 {
377 durStr = monitorCmd.Arg(0)
378 } else if len(cmdArgs) > 0 {
379 durStr = cmdArgs[0]
380 }
381
382 if durStr == "" {
383 _, _ = fmt.Fprintln(cmd.sesh, "Usage: monitor <topic> <duration>")
384 return fmt.Errorf("duration is required")
385 }
386
387 dur, err := time.ParseDuration(durStr)
388 if err != nil {
389 return fmt.Errorf("invalid duration %q: %w", durStr, err)
390 }
391
392 winEnd := time.Now().UTC().Add(dur)
393 handler.Logger.Info(
394 "upserting pipe monitor",
395 "topic", resolvedTopic,
396 "dur", dur,
397 "window", winEnd.UTC().Format(time.RFC3339),
398 )
399 err = handler.DBPool.UpsertPipeMonitor(user.ID, resolvedTopic, dur, &winEnd)
400 if err != nil {
401 return fmt.Errorf("failed to create monitor: %w", err)
402 }
403
404 _, _ = fmt.Fprintf(cmd.sesh, "monitor created: %s (window: %s)\r\n", resolvedTopic, dur)
405 return nil
406}
407
408func (handler *CliHandler) status(cmd *CliCmd, user *db.User) error {
409 if user == nil {
410 return fmt.Errorf("access denied")
411 }
412
413 monitors, err := handler.DBPool.FindPipeMonitorsByUser(user.ID)
414 if err != nil {
415 return fmt.Errorf("failed to fetch monitors: %w", err)
416 }
417
418 if len(monitors) == 0 {
419 _, _ = fmt.Fprintln(cmd.sesh, "no monitors found")
420 return nil
421 }
422
423 writer := tabwriter.NewWriter(cmd.sesh, 0, 0, 2, ' ', tabwriter.TabIndent)
424 _, _ = fmt.Fprintln(writer, "Topic\tStatus\tWindow\tLast Ping\tWindow End\tReason")
425
426 for _, m := range monitors {
427 status := "healthy"
428 reason := ""
429 if err := m.Status(); err != nil {
430 status = "unhealthy"
431 reason = err.Error()
432 }
433
434 lastPing := "never"
435 if m.LastPing != nil {
436 lastPing = m.LastPing.UTC().Format(time.RFC3339)
437 }
438
439 windowEnd := ""
440 if m.WindowEnd != nil {
441 windowEnd = m.WindowEnd.UTC().Format(time.RFC3339)
442 }
443
444 _, _ = fmt.Fprintf(
445 writer,
446 "%s\t%s\t%s\t%s\t%s\t%s\r\n",
447 m.Topic,
448 status,
449 m.WindowDur.String(),
450 lastPing,
451 windowEnd,
452 reason,
453 )
454 }
455 _ = writer.Flush()
456 return nil
457}
458
459func (handler *CliHandler) uptime(cmd *CliCmd, topic string, user *db.User) error {
460 if user == nil {
461 return fmt.Errorf("access denied")
462 }
463
464 if topic == "" {
465 _, _ = fmt.Fprintln(cmd.sesh, "usage: uptime <topic> [--from <time>] [--to <time>]")
466 _, _ = fmt.Fprintln(cmd.sesh, " --from: start time (RFC3339 or duration like '24h', '7d', default: 24h)")
467 _, _ = fmt.Fprintln(cmd.sesh, " --to: end time (RFC3339, default: now)")
468 return nil
469 }
470
471 fs := flag.NewFlagSet("uptime", flag.ContinueOnError)
472 fs.SetOutput(cmd.sesh)
473 fromStr := fs.String("from", "", "start time (RFC3339 or duration like '24h', '7d')")
474 toStr := fs.String("to", "", "end time (RFC3339, defaults to now)")
475
476 if err := fs.Parse(cmd.args); err != nil {
477 return nil
478 }
479
480 topicResult := resolveTopic(TopicResolveInput{
481 UserName: cmd.userName,
482 Topic: topic,
483 IsAdmin: cmd.isAdmin,
484 IsPublic: false,
485 })
486 resolvedTopic := topicResult.Name
487
488 monitor, err := handler.DBPool.FindPipeMonitorByTopic(user.ID, resolvedTopic)
489 if err != nil {
490 if errors.Is(err, sql.ErrNoRows) {
491 return fmt.Errorf("monitor not found: %s", topic)
492 }
493 return fmt.Errorf("failed to find monitor: %w", err)
494 }
495
496 now := time.Now().UTC()
497 to := now
498 from := now.Add(-24 * time.Hour)
499
500 if *fromStr != "" {
501 if parsed, err := time.Parse(time.RFC3339, *fromStr); err == nil {
502 from = parsed.UTC()
503 } else if dur, err := parseDuration(*fromStr); err == nil {
504 from = now.Add(-dur)
505 } else {
506 return fmt.Errorf("invalid --from value: %s", *fromStr)
507 }
508 }
509
510 if *toStr != "" {
511 if parsed, err := time.Parse(time.RFC3339, *toStr); err == nil {
512 to = parsed.UTC()
513 } else {
514 return fmt.Errorf("invalid --to value: %s", *toStr)
515 }
516 }
517
518 history, err := handler.DBPool.FindPipeMonitorHistory(monitor.ID, from, to)
519 if err != nil {
520 return fmt.Errorf("failed to fetch history: %w", err)
521 }
522
523 result := db.ComputeUptime(history, from, to)
524
525 _, _ = fmt.Fprintf(cmd.sesh, "Monitor: %s\r\n", topic)
526 _, _ = fmt.Fprintf(cmd.sesh, "Period: %s to %s\r\n", from.Format(time.RFC3339), to.Format(time.RFC3339))
527 _, _ = fmt.Fprintf(cmd.sesh, "Total Duration: %s\r\n", result.TotalDuration.Round(time.Second))
528 _, _ = fmt.Fprintf(cmd.sesh, "Uptime Duration: %s\r\n", result.UptimeDuration.Round(time.Second))
529 _, _ = fmt.Fprintf(cmd.sesh, "Uptime: %.2f%%\r\n", result.UptimePercent)
530
531 return nil
532}
533
534func parseDuration(s string) (time.Duration, error) {
535 if len(s) == 0 {
536 return 0, fmt.Errorf("empty duration")
537 }
538 last := s[len(s)-1]
539 if last == 'd' {
540 var n int
541 _, err := fmt.Sscanf(s, "%d", &n)
542 if err != nil {
543 return 0, fmt.Errorf("invalid duration: %s", s)
544 }
545 return time.Duration(n) * 24 * time.Hour, nil
546 }
547 return time.ParseDuration(s)
548}
549
550func MonitorRss(dbpool DB, user *db.User, domain string) (string, error) {
551 if user == nil {
552 return "", fmt.Errorf("access denied")
553 }
554
555 monitors, err := dbpool.FindPipeMonitorsByUser(user.ID)
556 if err != nil {
557 return "", fmt.Errorf("failed to fetch monitors: %w", err)
558 }
559
560 now := time.Now()
561 feed := &feeds.Feed{
562 Title: fmt.Sprintf("Pipe Monitors for %s", user.Name),
563 Link: &feeds.Link{Href: fmt.Sprintf("https://%s", domain)},
564 Description: "Alerts for pipe monitor status changes",
565 Author: &feeds.Author{Name: user.Name},
566 Created: now,
567 }
568
569 var feedItems []*feeds.Item
570 for _, m := range monitors {
571 if err := m.Status(); err != nil {
572 item := &feeds.Item{
573 Id: fmt.Sprintf("%s-%s-%d", user.ID, m.Topic, now.Unix()),
574 Title: fmt.Sprintf("ALERT: %s is unhealthy", m.Topic),
575 Link: &feeds.Link{Href: fmt.Sprintf("https://%s", domain)},
576 Description: err.Error(),
577 Created: now,
578 Updated: now,
579 Author: &feeds.Author{Name: user.Name},
580 }
581 feedItems = append(feedItems, item)
582 }
583 }
584 feed.Items = feedItems
585
586 rss, err := feed.ToRss()
587 if err != nil {
588 return "", fmt.Errorf("failed to generate RSS: %w", err)
589 }
590
591 return rss, nil
592}
593
594func (handler *CliHandler) pub(cmd *CliCmd, topic string, clientID string) error {
595 pubCmd := flagSet("pub", cmd.sesh)
596 access := pubCmd.String("a", "", "Comma separated list of pico usernames or ssh-key fingerprints to allow access to a topic")
597 empty := pubCmd.Bool("e", false, "Send an empty message to subs")
598 public := pubCmd.Bool("p", false, "Publish message to public topic")
599 block := pubCmd.Bool("b", true, "Block writes until a subscriber is available")
600 timeout := pubCmd.Duration("t", 30*24*time.Hour, "Timeout as a Go duration to block for a subscriber to be available. Valid time units are 'ns', 'us' (or 'µs'), 'ms', 's', 'm', 'h'. Default is 30 days.")
601 clean := pubCmd.Bool("c", false, "Don't send status messages")
602 dispatcher := pubCmd.String("d", "multicast", "Type of dispatcher (e.g. multicast, round_robin)")
603
604 if !flagCheck(pubCmd, topic, cmd.args) {
605 return fmt.Errorf("invalid cmd args")
606 }
607
608 if pubCmd.NArg() == 1 && topic == "" {
609 topic = pubCmd.Arg(0)
610 }
611
612 handler.Logger.Info(
613 "flags parsed",
614 "cmd", "pub",
615 "empty", *empty,
616 "public", *public,
617 "block", *block,
618 "timeout", *timeout,
619 "topic", topic,
620 "access", *access,
621 "clean", *clean,
622 "dispatcher", *dispatcher,
623 )
624
625 var accessList []string
626
627 if *access != "" {
628 accessList = parseArgList(*access)
629 }
630
631 var rw io.ReadWriter
632 if *empty {
633 rw = bytes.NewBuffer(make([]byte, 1))
634 } else {
635 rw = cmd.sesh
636 }
637
638 if topic == "" {
639 topic = uuid.NewString()
640 }
641
642 msgFlag := ""
643 if *public {
644 msgFlag = "-p "
645 }
646
647 // Initial resolution to get the topic name for access storage
648 initialResult := resolveTopic(TopicResolveInput{
649 UserName: cmd.userName,
650 Topic: topic,
651 IsAdmin: cmd.isAdmin,
652 IsPublic: *public,
653 })
654 name := initialResult.Name
655
656 var accessListCreator bool
657 _, loaded := handler.Access.LoadOrStore(name, accessList)
658 if !loaded {
659 defer func() {
660 handler.Access.Delete(name)
661 }()
662 accessListCreator = true
663 }
664
665 // Check for existing access list and resolve final topic name
666 existingAccessList, hasExistingAccess := handler.Access.Load(initialResult.WithoutUser)
667 result := resolveTopic(TopicResolveInput{
668 UserName: cmd.userName,
669 Topic: topic,
670 IsAdmin: cmd.isAdmin,
671 IsPublic: *public,
672 ExistingAccessList: existingAccessList,
673 HasExistingAccess: hasExistingAccess,
674 IsAccessCreator: accessListCreator,
675 HasUserAccess: checkAccess(existingAccessList, cmd.userName, cmd.sesh),
676 })
677 name = result.Name
678
679 if result.GenerateNewTopic {
680 topic = uuid.NewString()
681 name = toPublicTopic(topic)
682 }
683
684 if !*clean {
685 fmtTopic := topic
686 if *access != "" {
687 fmtTopic = fmt.Sprintf("%s/%s", cmd.userName, topic)
688 }
689
690 _, _ = fmt.Fprintf(
691 cmd.sesh,
692 "subscribe to this channel:\n ssh %s sub %s%s\n",
693 toSshCmd(handler.Cfg),
694 msgFlag,
695 fmtTopic,
696 )
697 }
698
699 if *block {
700 count := 0
701 for topic, channel := range handler.PubSub.GetChannels() {
702 if topic == name || (psub.HasWildcard(topic) && psub.MatchTopic(topic, name)) {
703 for _, client := range channel.GetClients() {
704 if client.Direction == psub.ChannelDirectionOutput || client.Direction == psub.ChannelDirectionInputOutput {
705 count++
706 }
707 }
708 if topic == name {
709 break
710 }
711 }
712 }
713
714 tt := *timeout
715 if count == 0 {
716 currentWaiters, _ := handler.Waiters.LoadOrStore(name, nil)
717 handler.Waiters.Store(name, append(currentWaiters, clientID))
718
719 termMsg := "no subs found ... waiting"
720 if tt > 0 {
721 termMsg += " " + tt.String()
722 }
723
724 if !*clean {
725 _, _ = fmt.Fprintln(cmd.sesh, termMsg)
726 }
727
728 ready := make(chan struct{})
729
730 go func() {
731 for {
732 select {
733 case <-cmd.pipeCtx.Done():
734 cmd.cancel()
735 return
736 case <-time.After(1 * time.Millisecond):
737 count := 0
738 for topic, channel := range handler.PubSub.GetChannels() {
739 if topic == name || (psub.HasWildcard(topic) && psub.MatchTopic(topic, name)) {
740 for _, client := range channel.GetClients() {
741 if client.Direction == psub.ChannelDirectionOutput || client.Direction == psub.ChannelDirectionInputOutput {
742 count++
743 }
744 }
745 if topic == name {
746 break
747 }
748 }
749 }
750
751 if count > 0 {
752 close(ready)
753 return
754 }
755 }
756 }
757 }()
758
759 select {
760 case <-ready:
761 case <-cmd.pipeCtx.Done():
762 case <-time.After(tt):
763 cmd.cancel()
764
765 if !*clean {
766 return fmt.Errorf("timeout reached, exiting")
767 } else {
768 err := cmd.sesh.Exit(1)
769 if err != nil {
770 handler.Logger.Error("error exiting session", "err", err)
771 }
772
773 _ = cmd.sesh.Close()
774 }
775 }
776
777 newWaiters, _ := handler.Waiters.LoadOrStore(name, nil)
778 newWaiters = slices.DeleteFunc(newWaiters, func(cl string) bool {
779 return cl == clientID
780 })
781 handler.Waiters.Store(name, newWaiters)
782
783 var toDelete []string
784
785 for channel, clients := range handler.Waiters.Range {
786 if len(clients) == 0 {
787 toDelete = append(toDelete, channel)
788 }
789 }
790
791 for _, channel := range toDelete {
792 handler.Waiters.Delete(channel)
793 }
794 }
795 }
796
797 if !*clean {
798 _, _ = fmt.Fprintln(cmd.sesh, "sending msg ...")
799 }
800
801 throttledRW := newThrottledMonitorRW(rw, handler, cmd, name)
802
803 var dsp psub.MessageDispatcher
804 dsp = &psub.MulticastDispatcher{}
805 if *dispatcher == "round_robin" {
806 dsp = &psub.RoundRobinDispatcher{}
807 }
808 channel := psub.NewChannel(name)
809 _ = handler.PubSub.SetDispatcher(dsp, []*psub.Channel{channel})
810
811 err := handler.PubSub.Pub(
812 cmd.pipeCtx,
813 clientID,
814 throttledRW,
815 []*psub.Channel{channel},
816 *block,
817 )
818
819 if !*clean {
820 _, _ = fmt.Fprintln(cmd.sesh, "msg sent!")
821 }
822
823 if err != nil && !*clean {
824 return err
825 }
826
827 handler.updateMonitor(cmd, name)
828
829 return nil
830}
831
832func (handler *CliHandler) updateMonitor(cmd *CliCmd, topic string) {
833 if cmd.user == nil {
834 return
835 }
836
837 handler.Logger.Info("update monitor", "topic", topic)
838 monitor, err := handler.DBPool.FindPipeMonitorByTopic(cmd.user.ID, topic)
839 if err != nil || monitor == nil {
840 handler.Logger.Info("no monitor found", "topic", topic)
841 return
842 }
843
844 now := time.Now().UTC()
845
846 // Fixed window semantics: windows are discrete, non-overlapping time slots.
847 // - last_ping: always updated to show most recent activity (user visibility)
848 // - window_end: only advances when current time exceeds it (health scheduling)
849
850 // If we're past the current window, advance to the window containing `now`
851 newWindowEnd := *monitor.WindowEnd
852 if !now.Before(*monitor.WindowEnd) {
853 // Record history for the completed window before advancing
854 // This captures that the old window was healthy (had activity)
855 if err := handler.DBPool.InsertPipeMonitorHistory(monitor.ID, monitor.WindowDur, monitor.WindowEnd, monitor.LastPing); err != nil {
856 handler.Logger.Error("failed to insert monitor history", "err", err, "topic", topic)
857 }
858
859 // Calculate which window period `now` falls into
860 elapsed := now.Sub(*monitor.WindowEnd)
861 periods := int(elapsed/monitor.WindowDur) + 1
862 newWindowEnd = monitor.WindowEnd.Add(time.Duration(periods) * monitor.WindowDur)
863
864 if err := handler.DBPool.UpsertPipeMonitor(cmd.user.ID, topic, monitor.WindowDur, &newWindowEnd); err != nil {
865 handler.Logger.Error("failed to advance monitor window", "err", err, "topic", topic)
866 }
867 handler.Logger.Info("advanced monitor window",
868 "topic", topic,
869 "oldWindowEnd", monitor.WindowEnd.Format(time.RFC3339),
870 "newWindowEnd", newWindowEnd.Format(time.RFC3339),
871 "periodsMissed", periods-1,
872 )
873 }
874
875 // Always record the latest ping for user visibility
876 if err := handler.DBPool.UpdatePipeMonitorLastPing(cmd.user.ID, topic, &now); err != nil {
877 handler.Logger.Error("failed to update monitor last_ping", "err", err, "topic", topic)
878 }
879
880 handler.Logger.Info("recorded monitor ping",
881 "topic", topic,
882 "pingTime", now.Format(time.RFC3339),
883 "windowEnd", newWindowEnd.Format(time.RFC3339),
884 )
885}
886
887const monitorThrottleInterval = 15 * time.Second
888
889type throttledMonitorRW struct {
890 rw io.ReadWriter
891 handler *CliHandler
892 cmd *CliCmd
893 topic string
894 lastPing atomic.Int64 // Unix nanoseconds
895}
896
897func newThrottledMonitorRW(rw io.ReadWriter, handler *CliHandler, cmd *CliCmd, topic string) *throttledMonitorRW {
898 return &throttledMonitorRW{
899 rw: rw,
900 handler: handler,
901 cmd: cmd,
902 topic: topic,
903 }
904}
905
906func (t *throttledMonitorRW) throttledUpdate() {
907 now := time.Now().UnixNano()
908 last := t.lastPing.Load()
909
910 // First ping (last == 0) or interval elapsed
911 if last == 0 || now-last >= int64(monitorThrottleInterval) {
912 if t.lastPing.CompareAndSwap(last, now) {
913 t.handler.updateMonitor(t.cmd, t.topic)
914 }
915 }
916}
917
918func (t *throttledMonitorRW) Read(p []byte) (int, error) {
919 n, err := t.rw.Read(p)
920 if n > 0 {
921 t.throttledUpdate()
922 }
923 return n, err
924}
925
926func (t *throttledMonitorRW) Write(p []byte) (int, error) {
927 n, err := t.rw.Write(p)
928 if n > 0 {
929 t.throttledUpdate()
930 }
931 return n, err
932}
933
934func (t *throttledMonitorRW) Close() error {
935 if closer, ok := t.rw.(io.Closer); ok {
936 return closer.Close()
937 }
938 return nil
939}
940
941func (handler *CliHandler) sub(cmd *CliCmd, topic string, clientID string) error {
942 subCmd := flagSet("sub", cmd.sesh)
943 access := subCmd.String("a", "", "Comma separated list of pico usernames or ssh-key fingerprints to allow access to a topic")
944 public := subCmd.Bool("p", false, "Subscribe to a public topic")
945 keepAlive := subCmd.Bool("k", false, "Keep the subscription alive even after the publisher has died")
946 clean := subCmd.Bool("c", false, "Don't send status messages")
947
948 if !flagCheck(subCmd, topic, cmd.args) {
949 return fmt.Errorf("invalid cmd args")
950 }
951
952 if subCmd.NArg() == 1 && topic == "" {
953 topic = subCmd.Arg(0)
954 }
955
956 handler.Logger.Info(
957 "flags parsed",
958 "cmd", cmd,
959 "public", *public,
960 "keepAlive", *keepAlive,
961 "topic", topic,
962 "clean", *clean,
963 "access", *access,
964 )
965
966 var accessList []string
967
968 if *access != "" {
969 accessList = parseArgList(*access)
970 }
971
972 // Initial resolution to get the topic name for access storage
973 initialResult := resolveTopic(TopicResolveInput{
974 UserName: cmd.userName,
975 Topic: topic,
976 IsAdmin: cmd.isAdmin,
977 IsPublic: *public,
978 })
979 name := initialResult.Name
980
981 var accessListCreator bool
982
983 _, loaded := handler.Access.LoadOrStore(name, accessList)
984 if !loaded {
985 defer func() {
986 handler.Access.Delete(name)
987 }()
988 accessListCreator = true
989 }
990
991 // Check for existing access list and resolve final topic name
992 existingAccessList, hasExistingAccess := handler.Access.Load(initialResult.WithoutUser)
993 result := resolveTopic(TopicResolveInput{
994 UserName: cmd.userName,
995 Topic: topic,
996 IsAdmin: cmd.isAdmin,
997 IsPublic: *public,
998 ExistingAccessList: existingAccessList,
999 HasExistingAccess: hasExistingAccess,
1000 IsAccessCreator: accessListCreator,
1001 HasUserAccess: checkAccess(existingAccessList, cmd.userName, cmd.sesh),
1002 })
1003 name = result.Name
1004
1005 if result.AccessDenied {
1006 return fmt.Errorf("access denied")
1007 }
1008
1009 err := handler.PubSub.Sub(
1010 cmd.pipeCtx,
1011 clientID,
1012 cmd.sesh,
1013 []*psub.Channel{
1014 psub.NewChannel(name),
1015 },
1016 *keepAlive,
1017 )
1018
1019 if err != nil && !*clean {
1020 return err
1021 }
1022
1023 return nil
1024}
1025
1026func (handler *CliHandler) pipe(cmd *CliCmd, topic string, clientID string) error {
1027 pipeCmd := flagSet("pipe", cmd.sesh)
1028 access := pipeCmd.String("a", "", "Comma separated list of pico usernames or ssh-key fingerprints to allow access to a topic")
1029 public := pipeCmd.Bool("p", false, "Pipe to a public topic")
1030 replay := pipeCmd.Bool("r", false, "Replay messages to the client that sent it")
1031 clean := pipeCmd.Bool("c", false, "Don't send status messages")
1032 dispatcher := pipeCmd.String("d", "multicast", "Type of dispatcher (e.g. multicast, round_robin)")
1033
1034 if !flagCheck(pipeCmd, topic, cmd.args) {
1035 return fmt.Errorf("invalid cmd args")
1036 }
1037
1038 if pipeCmd.NArg() == 1 && topic == "" {
1039 topic = pipeCmd.Arg(0)
1040 }
1041
1042 handler.Logger.Info(
1043 "flags parsed",
1044 "cmd", cmd,
1045 "public", *public,
1046 "replay", *replay,
1047 "topic", topic,
1048 "access", *access,
1049 "clean", *clean,
1050 "dispatcher", *dispatcher,
1051 )
1052
1053 var accessList []string
1054
1055 if *access != "" {
1056 accessList = parseArgList(*access)
1057 }
1058
1059 isCreator := topic == ""
1060 if isCreator {
1061 topic = uuid.NewString()
1062 }
1063
1064 flagMsg := ""
1065 if *public {
1066 flagMsg = "-p "
1067 }
1068
1069 // Initial resolution to get the topic name for access storage
1070 initialResult := resolveTopic(TopicResolveInput{
1071 UserName: cmd.userName,
1072 Topic: topic,
1073 IsAdmin: cmd.isAdmin,
1074 IsPublic: *public,
1075 })
1076 name := initialResult.Name
1077
1078 var accessListCreator bool
1079
1080 _, loaded := handler.Access.LoadOrStore(name, accessList)
1081 if !loaded {
1082 defer func() {
1083 handler.Access.Delete(name)
1084 }()
1085 accessListCreator = true
1086 }
1087
1088 // Check for existing access list and resolve final topic name
1089 existingAccessList, hasExistingAccess := handler.Access.Load(initialResult.WithoutUser)
1090 result := resolveTopic(TopicResolveInput{
1091 UserName: cmd.userName,
1092 Topic: topic,
1093 IsAdmin: cmd.isAdmin,
1094 IsPublic: *public,
1095 ExistingAccessList: existingAccessList,
1096 HasExistingAccess: hasExistingAccess,
1097 IsAccessCreator: accessListCreator,
1098 HasUserAccess: checkAccess(existingAccessList, cmd.userName, cmd.sesh),
1099 })
1100 name = result.Name
1101
1102 if result.GenerateNewTopic {
1103 topic = uuid.NewString()
1104 name = toPublicTopic(topic)
1105 }
1106
1107 if isCreator && !*clean {
1108 fmtTopic := topic
1109 if *access != "" {
1110 fmtTopic = fmt.Sprintf("%s/%s", cmd.userName, topic)
1111 }
1112
1113 _, _ = fmt.Fprintf(
1114 cmd.sesh,
1115 "subscribe to this topic:\n ssh %s sub %s%s\n",
1116 toSshCmd(handler.Cfg),
1117 flagMsg,
1118 fmtTopic,
1119 )
1120 }
1121
1122 throttledRW := newThrottledMonitorRW(cmd.sesh, handler, cmd, name)
1123
1124 var dsp psub.MessageDispatcher
1125 dsp = &psub.MulticastDispatcher{}
1126 if *dispatcher == "round_robin" {
1127 dsp = &psub.RoundRobinDispatcher{}
1128 }
1129 channel := psub.NewChannel(name)
1130 _ = handler.PubSub.SetDispatcher(dsp, []*psub.Channel{channel})
1131
1132 readErr, writeErr := handler.PubSub.Pipe(
1133 cmd.pipeCtx,
1134 clientID,
1135 throttledRW,
1136 []*psub.Channel{
1137 psub.NewChannel(name),
1138 },
1139 *replay,
1140 )
1141
1142 if readErr != nil && !*clean {
1143 return readErr
1144 }
1145
1146 if writeErr != nil && !*clean {
1147 return writeErr
1148 }
1149
1150 handler.updateMonitor(cmd, name)
1151
1152 return nil
1153}
1154
1155func toSshCmd(cfg *Config) string {
1156 port := ""
1157 if cfg.PortOverride != "22" {
1158 port = fmt.Sprintf("-p %s ", cfg.PortOverride)
1159 }
1160 return fmt.Sprintf("%s%s", port, cfg.Domain)
1161}
1162
1163// parseArgList parses a comma separated list of arguments.
1164func parseArgList(arg string) []string {
1165 argList := strings.Split(arg, ",")
1166 for i, acc := range argList {
1167 argList[i] = strings.TrimSpace(acc)
1168 }
1169 return argList
1170}
1171
1172// checkAccess checks if the user has access to a topic based on an access list.
1173func checkAccess(accessList []string, userName string, sesh *pssh.SSHServerConnSession) bool {
1174 for _, acc := range accessList {
1175 if acc == userName {
1176 return true
1177 }
1178
1179 if key := sesh.PublicKey(); key != nil && acc == gossh.FingerprintSHA256(key) {
1180 return true
1181 }
1182 }
1183
1184 return false
1185}
1186
1187func flagSet(cmdName string, sesh *pssh.SSHServerConnSession) *flag.FlagSet {
1188 cmd := flag.NewFlagSet(cmdName, flag.ContinueOnError)
1189 cmd.SetOutput(sesh)
1190 cmd.Usage = func() {
1191 _, _ = fmt.Fprintf(cmd.Output(), "Usage: %s <topic> [args...]\nArgs:\n", cmdName)
1192 cmd.PrintDefaults()
1193 }
1194 return cmd
1195}
1196
1197func flagCheck(cmd *flag.FlagSet, posArg string, cmdArgs []string) bool {
1198 err := cmd.Parse(cmdArgs)
1199
1200 if err != nil || posArg == "help" {
1201 if posArg == "help" {
1202 cmd.Usage()
1203 }
1204 return false
1205 }
1206 return true
1207}
1208
1209func clientInfo(clients []*psub.Client, isAdmin bool, clientType string) string {
1210 if len(clients) == 0 {
1211 return ""
1212 }
1213
1214 outputData := fmt.Sprintf(" %s:\r\n", clientType)
1215
1216 for _, client := range clients {
1217 if strings.HasPrefix(client.ID, "admin-") && !isAdmin {
1218 continue
1219 }
1220
1221 outputData += fmt.Sprintf(" - %s\r\n", client.ID)
1222 }
1223
1224 return outputData
1225}