dashboard / erock/pico / feat(tui.tuns): conn events #54 rss

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

diff --git a/Makefile b/Makefile
index 01f00a4..d452d2f 100644
--- a/Makefile
+++ b/Makefile
@@ -126,10 +126,11 @@ migrate:
 	$(DOCKER_CMD) exec -i $(DB_CONTAINER) psql -U $(PGUSER) -d $(PGDATABASE) < ./sql/migrations/20241114_add_namespace_to_analytics.sql
 	$(DOCKER_CMD) exec -i $(DB_CONTAINER) psql -U $(PGUSER) -d $(PGDATABASE) < ./sql/migrations/20241125_add_content_type_to_analytics.sql
 	$(DOCKER_CMD) exec -i $(DB_CONTAINER) psql -U $(PGUSER) -d $(PGDATABASE) < ./sql/migrations/20241202_add_more_idx_analytics.sql
+	$(DOCKER_CMD) exec -i $(DB_CONTAINER) psql -U $(PGUSER) -d $(PGDATABASE) < ./sql/migrations/20250319_add_tuns_event_logs_table.sql
 .PHONY: migrate
 
 latest:
-	$(DOCKER_CMD) exec -i $(DB_CONTAINER) psql -U $(PGUSER) -d $(PGDATABASE) < ./sql/migrations/20241202_add_more_idx_analytics.sql
+	$(DOCKER_CMD) exec -i $(DB_CONTAINER) psql -U $(PGUSER) -d $(PGDATABASE) < ./sql/migrations/20250319_add_tuns_event_logs_table.sql
 .PHONY: latest
 
 psql:

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

diff --git a/pkg/apps/auth/api.go b/pkg/apps/auth/api.go
index bb72784..6534040 100644
--- a/pkg/apps/auth/api.go
+++ b/pkg/apps/auth/api.go
@@ -20,6 +20,7 @@ import (
 	"github.com/picosh/pico/pkg/db/postgres"
 	"github.com/picosh/pico/pkg/shared"
 	"github.com/picosh/utils"
+	"github.com/picosh/utils/pipe"
 	"github.com/picosh/utils/pipe/metrics"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 )
@@ -723,6 +724,39 @@ func metricDrainSub(ctx context.Context, dbpool db.DB, logger *slog.Logger, secr
 	}
 }
 
+func tunsEventLogDrainSub(ctx context.Context, dbpool db.DB, logger *slog.Logger, secret string) {
+	drain := pipe.NewReconnectReadWriteCloser(
+		ctx,
+		logger,
+		shared.NewPicoPipeClient(),
+		"tuns-event-drain-sub",
+		"sub tuns-event-drain -k",
+		100,
+		10*time.Millisecond,
+	)
+
+	for {
+		scanner := bufio.NewScanner(drain)
+		scanner.Buffer(make([]byte, 32*1024), 32*1024)
+		for scanner.Scan() {
+			line := scanner.Text()
+			clean := strings.TrimSpace(line)
+			var log db.TunsEventLog
+			err := json.Unmarshal([]byte(clean), &log)
+			if err != nil {
+				logger.Error("could not unmarshal line", "err", err)
+				continue
+			}
+
+			logger.Info("inserting tuns event log", "log", log)
+			err = dbpool.InsertTunsEventLog(&log)
+			if err != nil {
+				logger.Error("could not insert tuns event log", "err", err)
+			}
+		}
+	}
+}
+
 func authMux(apiConfig *shared.ApiConfig) *http.ServeMux {
 	serverRoot, err := fs.Sub(embedFS, "public")
 	if err != nil {
@@ -792,6 +826,7 @@ func StartApiServer() {
 
 	// gather metrics in the auth service
 	go metricDrainSub(ctx, db, logger, cfg.Secret)
+	go tunsEventLogDrainSub(ctx, db, logger, cfg.Secret)
 
 	defer ctx.Done()
 

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

diff --git a/pkg/db/db.go b/pkg/db/db.go
index f251d88..56acce0 100644
--- a/pkg/db/db.go
+++ b/pkg/db/db.go
@@ -323,6 +323,20 @@ type UserServiceStats struct {
 	LatestUpdatedAt  time.Time
 }
 
+type TunsEventLog struct {
+	ID             string     `json:"id"`
+	ServerID       string     `json:"server_id"`
+	Time           *time.Time `json:"time"`
+	User           string     `json:"user"`
+	UserId         string     `json:"user_id"`
+	RemoteAddr     string     `json:"remote_addr"`
+	EventType      string     `json:"event_type"`
+	TunnelType     string     `json:"tunnel_type"`
+	ConnectionType string     `json:"connection_type"`
+	TunnelAddrs    []string   `json:"tunnel_addrs"`
+	CreatedAt      *time.Time `json:"created_at"`
+}
+
 var NameValidator = regexp.MustCompile("^[a-zA-Z0-9]{1,50}$")
 var DenyList = []string{
 	"admin",
@@ -415,5 +429,8 @@ type DB interface {
 
 	FindUserStats(userID string) (*UserStats, error)
 
+	InsertTunsEventLog(log *TunsEventLog) error
+	FindTunsEventLog(userID, addr string) ([]*TunsEventLog, error)
+
 	Close() error
 }

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

diff --git a/pkg/db/postgres/storage.go b/pkg/db/postgres/storage.go
index ce39adc..d0eefe0 100644
--- a/pkg/db/postgres/storage.go
+++ b/pkg/db/postgres/storage.go
@@ -1796,6 +1796,45 @@ func (me *PsqlDB) findPagesStats(userID string) (*db.UserServiceStats, error) {
 	return &stats, nil
 }
 
+func (me *PsqlDB) InsertTunsEventLog(log *db.TunsEventLog) error {
+	_, err := me.Db.Exec(
+		`INSERT INTO tuns_event_logs
+			(user_id, server_id, remote_addr, tunnel_type, connection_type, tunnel_addrs)
+		VALUES
+			($1, $2, $3, $4, $5, $6)`,
+		log.UserId, log.ServerID, log.RemoteAddr, log.TunnelType, log.ConnectionType, log.TunnelAddrs,
+	)
+	return err
+}
+
+func (me *PsqlDB) FindTunsEventLog(userID, addr string) ([]*db.TunsEventLog, error) {
+	logs := []*db.TunsEventLog{}
+	rs, err := me.Db.Query(
+		`SELECT user_id, server_id, remote_addr, tunnel_type, connection_type, tunnel_addrs, created_at
+		FROM tuns_event_logs WHERE user_id=$1 AND tunnel_addrs @> ARRAY[$2] ORDER BY created_at DESC`, userID, addr)
+	if err != nil {
+		return nil, err
+	}
+
+	for rs.Next() {
+		log := db.TunsEventLog{}
+		err := rs.Scan(
+			&log.ID, &log.UserId, &log.ServerID, &log.RemoteAddr,
+			&log.TunnelType, &log.ConnectionType, &log.TunnelAddrs, &log.ConnectionType, &log.CreatedAt,
+		)
+		if err != nil {
+			return nil, err
+		}
+		logs = append(logs, &log)
+	}
+
+	if rs.Err() != nil {
+		return nil, rs.Err()
+	}
+
+	return logs, nil
+}
+
 func (me *PsqlDB) FindUserStats(userID string) (*db.UserStats, error) {
 	stats := db.UserStats{}
 	rs, err := me.Db.Query(`SELECT cur_space, count(id), min(created_at), max(created_at), max(updated_at) FROM posts WHERE user_id=$1 GROUP BY cur_space`, userID)

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

diff --git a/pkg/db/stub/stub.go b/pkg/db/stub/stub.go
index b1d8d2d..59df3cf 100644
--- a/pkg/db/stub/stub.go
+++ b/pkg/db/stub/stub.go
@@ -268,3 +268,11 @@ func (me *StubDB) FindTagsForUser(userID string, tag string) ([]string, error) {
 func (me *StubDB) FindUserStats(userID string) (*db.UserStats, error) {
 	return nil, notImpl
 }
+
+func (me *StubDB) InsertTunsEventLog(log *db.TunsEventLog) error {
+	return notImpl
+}
+
+func (me *StubDB) FindTunsEventLog(userID, addr string) ([]*db.TunsEventLog, error) {
+	return nil, notImpl
+}

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

diff --git a/pkg/tui/tuns.go b/pkg/tui/tuns.go
index 8224e09..a632ec4 100644
--- a/pkg/tui/tuns.go
+++ b/pkg/tui/tuns.go
@@ -15,6 +15,7 @@ import (
 	"git.sr.ht/~rockorager/vaxis/vxfw/list"
 	"git.sr.ht/~rockorager/vaxis/vxfw/richtext"
 	"git.sr.ht/~rockorager/vaxis/vxfw/text"
+	"github.com/picosh/pico/pkg/db"
 	"github.com/picosh/pico/pkg/shared"
 	"github.com/picosh/utils/pipe"
 )
@@ -78,21 +79,25 @@ type ResultLogLineLoaded struct {
 
 type TunsLoaded struct{}
 
+type EventLogsLoaded struct{}
+
 type TunsPage struct {
 	shared *SharedModel
 
-	loading   bool
-	err       error
-	tuns      []TunsClientSimple
-	selected  string
-	focus     string
-	leftPane  list.Dynamic
-	rightPane *Pager
-	logs      []*ResultLog
-	logList   list.Dynamic
-	ctx       context.Context
-	done      context.CancelFunc
-	isAdmin   bool
+	loading      bool
+	err          error
+	tuns         []TunsClientSimple
+	selected     string
+	focus        string
+	leftPane     list.Dynamic
+	rightPane    *Pager
+	logs         []*ResultLog
+	logList      list.Dynamic
+	ctx          context.Context
+	done         context.CancelFunc
+	isAdmin      bool
+	eventLogs    []*db.TunsEventLog
+	eventLogList list.Dynamic
 }
 
 func NewTunsPage(shrd *SharedModel) *TunsPage {
@@ -103,6 +108,7 @@ func NewTunsPage(shrd *SharedModel) *TunsPage {
 	}
 	m.leftPane = list.Dynamic{DrawCursor: true, Builder: m.getLeftWidget}
 	m.logList = list.Dynamic{DrawCursor: true, Builder: m.getLogWidget}
+	m.eventLogList = list.Dynamic{DrawCursor: true, Builder: m.getEventLogWidget}
 	ff, _ := shrd.Dbpool.FindFeatureForUser(m.shared.User.ID, "admin")
 	if ff != nil {
 		m.isAdmin = true
@@ -145,6 +151,24 @@ func (m *TunsPage) getLogWidget(i uint, cursor uint) vxfw.Widget {
 	return txt
 }
 
+func (m *TunsPage) getEventLogWidget(i uint, cursor uint) vxfw.Widget {
+	if int(i) >= len(m.tuns) {
+		return nil
+	}
+
+	log := m.eventLogs[i]
+	style := vaxis.Style{Foreground: green}
+	if log.EventType == "disconnect" {
+		style = vaxis.Style{Foreground: red}
+	}
+	txt := richtext.New([]vaxis.Segment{
+		{Text: log.CreatedAt.Format(time.RFC3339) + " "},
+		{Text: log.EventType, Style: style},
+	})
+	txt.Softwrap = false
+	return txt
+}
+
 func (m *TunsPage) connectToLogs() error {
 	ctx, cancel := context.WithCancel(m.shared.Session.Context())
 	m.ctx = ctx
@@ -208,6 +232,8 @@ func (m *TunsPage) HandleEvent(ev vaxis.Event, ph vxfw.EventPhase) (vxfw.Command
 			m.logList.SetCursor(uint(len(m.logs) - 1))
 		}
 		return vxfw.RedrawCmd{}, nil
+	case EventLogsLoaded:
+		return vxfw.RedrawCmd{}, nil
 	case TunsLoaded:
 		m.focus = "tuns"
 		return vxfw.BatchCmd([]vxfw.Command{
@@ -218,6 +244,8 @@ func (m *TunsPage) HandleEvent(ev vaxis.Event, ph vxfw.EventPhase) (vxfw.Command
 		if msg.Matches(vaxis.KeyEnter) {
 			m.selected = m.tuns[m.leftPane.Cursor()].TunAddress
 			m.logs = []*ResultLog{}
+			m.eventLogs = []*db.TunsEventLog{}
+			go m.fetchEventLogs()
 			return vxfw.RedrawCmd{}, nil
 		}
 		if msg.Matches(vaxis.KeyTab) {
@@ -325,7 +353,20 @@ func (m *TunsPage) Draw(ctx vxfw.DrawContext) (vxfw.Surface, error) {
 			Characters: vaxis.Characters,
 			Max: vxfw.Size{
 				Width:  uint16(rightPaneW) - 4,
-				Height: ctx.Max.Height - uint16(ah) - 3,
+				Height: 15,
+			},
+		})
+		rightSurf.AddChild(0, ah, surf)
+		ah += int(surf.Size.Height)
+
+		brd = NewBorder(&m.eventLogList)
+		brd.Label = "conn events"
+		m.focusBorder(brd)
+		surf, _ = brd.Draw(vxfw.DrawContext{
+			Characters: vaxis.Characters,
+			Max: vxfw.Size{
+				Width:  uint16(rightPaneW) - 4,
+				Height: 15,
 			},
 		})
 		rightSurf.AddChild(0, ah, surf)
@@ -376,6 +417,15 @@ func fetch(fqdn, auth string) (map[string]*TunsClient, error) {
 	return data.Clients, nil
 }
 
+func (m *TunsPage) fetchEventLogs() {
+	logs, err := m.shared.Dbpool.FindTunsEventLog(m.shared.User.ID, m.selected)
+	if err != nil {
+		m.err = err
+		return
+	}
+	m.eventLogs = logs
+}
+
 func (m *TunsPage) fetchTuns() {
 	tMap, err := fetch("tuns.sh", m.shared.Cfg.TunsSecret)
 	if err != nil {

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

diff --git a/sql/migrations/20250319_add_tuns_event_logs_table.sql b/sql/migrations/20250319_add_tuns_event_logs_table.sql
new file mode 100644
index 0000000..e43b2d2
--- /dev/null
+++ b/sql/migrations/20250319_add_tuns_event_logs_table.sql
@@ -0,0 +1,17 @@
+CREATE TABLE IF NOT EXISTS tuns_event_logs (
+  id uuid NOT NULL DEFAULT uuid_generate_v4(),
+  user_id uuid NOT NULL,
+  server_id text,
+  remote_addr text,
+  event_type text,
+  tunnel_type text,
+  connection_type text,
+  tunnel_addrs text[],
+  created_at timestamp without time zone NOT NULL DEFAULT NOW(),
+  CONSTRAINT tuns_event_logs_pkey PRIMARY KEY (id),
+  CONSTRAINT fk_tuns_event_logs_user
+    FOREIGN KEY(user_id)
+  REFERENCES app_users(id)
+  ON DELETE CASCADE
+  ON UPDATE CASCADE
+);