update

internal/collector/collector.go

@@ -18,6 +18,7 @@ import (
 	"git.tornberg.me/go-otel/internal/prometheus"
 	"git.tornberg.me/go-otel/internal/sourcemaps"
 	"git.tornberg.me/go-otel/internal/storage"
+	"git.tornberg.me/go-otel/internal/todos"
 	"git.tornberg.me/go-otel/internal/uaclass"
 	"git.tornberg.me/go-otel/internal/websocket"
 
@@ -43,6 +44,7 @@ type Collector struct {
 
 	ingestLimiter *ipRateLimiter
 	ingestKey     string
+	todoStore     todos.Store
 	sourceMapKey  string
 
 	notifyIssuesFn func()

internal/collector/todos.go

@@ -0,0 +1,118 @@
+package collector
+
+import (
+	"crypto/rand"
+	"encoding/hex"
+	"encoding/json"
+	"log"
+	"net/http"
+	"strconv"
+	"strings"
+	"time"
+
+	"git.tornberg.me/go-otel/internal/todos"
+)
+
+const (
+	maxTodoBody = 16 << 10
+	maxTodoText = 4000
+	maxTodoMeta = 500
+)
+
+// SetTodoStore enables the todo endpoints.
+func (c *Collector) SetTodoStore(s todos.Store) { c.todoStore = s }
+
+func clipTo(s string, n int) string {
+	s = strings.TrimSpace(s)
+	if len(s) > n {
+		return s[:n]
+	}
+	return s
+}
+
+type todoPayload struct {
+	Project   string `json:"project"`
+	Text      string `json:"text"`
+	Component string `json:"component"`
+	Path      string `json:"path"`
+	Hash      string `json:"hash"`
+	URL       string `json:"url"`
+	UserAgent string `json:"ua"`
+	Session   string `json:"session"`
+}
+
+// HandleTodoCreate ingests a todo posted by otel-todo.js. It is guarded by
+// the same ingest key and rate limit as the other RUM endpoints.
+func (c *Collector) HandleTodoCreate(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodPost {
+		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
+		return
+	}
+	if c.todoStore == nil {
+		http.Error(w, "todos not enabled", http.StatusServiceUnavailable)
+		return
+	}
+	if !c.checkIngest(w, r) {
+		return
+	}
+	var p todoPayload
+	if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, maxTodoBody)).Decode(&p); err != nil {
+		http.Error(w, "invalid todo payload", http.StatusBadRequest)
+		return
+	}
+	text := clipTo(p.Text, maxTodoText)
+	if text == "" {
+		http.Error(w, "text is required", http.StatusBadRequest)
+		return
+	}
+	project := clipTo(p.Project, maxTodoMeta)
+	if project == "" {
+		project = "frontend"
+	}
+	id := make([]byte, 8)
+	_, _ = rand.Read(id)
+	t := todos.Todo{
+		ID:        hex.EncodeToString(id),
+		Project:   project,
+		Text:      text,
+		Component: clipTo(p.Component, maxTodoMeta),
+		Path:      clipTo(p.Path, maxTodoMeta),
+		Hash:      clipTo(p.Hash, maxTodoMeta),
+		URL:       clipTo(p.URL, maxTodoMeta),
+		UserAgent: clipTo(p.UserAgent, maxTodoMeta),
+		Session:   clipTo(p.Session, maxTodoMeta),
+		CreatedAt: time.Now().UTC(),
+	}
+	if err := c.todoStore.Add(r.Context(), t); err != nil {
+		log.Printf("failed to store todo: %v", err)
+		http.Error(w, "failed to store todo", http.StatusInternalServerError)
+		return
+	}
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(http.StatusCreated)
+	_ = json.NewEncoder(w).Encode(t)
+}
+
+// HandleTodoList returns todos, newest first. ?project= filters by project
+// (omit for all); ?limit= defaults to 100, max 1000. Registered behind auth.
+func (c *Collector) HandleTodoList(w http.ResponseWriter, r *http.Request) {
+	if c.todoStore == nil {
+		http.Error(w, "todos not enabled", http.StatusServiceUnavailable)
+		return
+	}
+	limit := 100
+	if v, err := strconv.Atoi(r.URL.Query().Get("limit")); err == nil && v > 0 {
+		limit = min(v, 1000)
+	}
+	items, err := c.todoStore.List(r.Context(), r.URL.Query().Get("project"), limit)
+	if err != nil {
+		log.Printf("failed to list todos: %v", err)
+		http.Error(w, "failed to list todos", http.StatusInternalServerError)
+		return
+	}
+	if items == nil {
+		items = []todos.Todo{}
+	}
+	w.Header().Set("Content-Type", "application/json")
+	_ = json.NewEncoder(w).Encode(items)
+}

internal/collector/todos_test.go

@@ -0,0 +1,45 @@
+package collector
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"git.tornberg.me/go-otel/internal/todos"
+)
+
+func TestTodoCreateAndList(t *testing.T) {
+	coll := New(nil, 10, nil)
+	st, err := todos.NewFileStore(filepath.Join(t.TempDir(), "todos.jsonl"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	coll.SetTodoStore(st)
+
+	post := func(body string) *httptest.ResponseRecorder {
+		w := httptest.NewRecorder()
+		coll.HandleTodoCreate(w, httptest.NewRequest(http.MethodPost, "/v1/todos", strings.NewReader(body)))
+		return w
+	}
+	if w := post(`{"project":"a","text":"fix cart","component":"Cart","path":"/cart","hash":"#x"}`); w.Code != http.StatusCreated {
+		t.Fatalf("create: %d %s", w.Code, w.Body)
+	}
+	post(`{"project":"b","text":"other"}`)
+	if w := post(`{"project":"a","text":"  "}`); w.Code != http.StatusBadRequest {
+		t.Fatalf("empty text: %d", w.Code)
+	}
+
+	list := func(q string) string {
+		w := httptest.NewRecorder()
+		coll.HandleTodoList(w, httptest.NewRequest(http.MethodGet, "/api/todos"+q, nil))
+		return w.Body.String()
+	}
+	if got := list("?project=a"); !strings.Contains(got, "fix cart") || strings.Contains(got, "other") || !strings.Contains(got, `"hash":"#x"`) {
+		t.Fatalf("project filter: %s", got)
+	}
+	if got := list(""); !strings.Contains(got, "fix cart") || !strings.Contains(got, "other") {
+		t.Fatalf("all: %s", got)
+	}
+}

internal/todos/todos.go

@@ -0,0 +1,155 @@
+// Package todos stores todo/issue notes filed from the browser by the
+// otel-todo.js widget.
+package todos
+
+import (
+	"bufio"
+	"context"
+	"encoding/json"
+	"fmt"
+	"os"
+	"sort"
+	"sync"
+	"time"
+
+	"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
+)
+
+// Todo is one note filed from a page. Project is the RUM service name.
+type Todo struct {
+	ID        string    `json:"id"`
+	Project   string    `json:"project"`
+	Text      string    `json:"text"`
+	Component string    `json:"component,omitempty"`
+	Path      string    `json:"path,omitempty"`
+	Hash      string    `json:"hash,omitempty"`
+	URL       string    `json:"url,omitempty"`
+	UserAgent string    `json:"userAgent,omitempty"`
+	Session   string    `json:"session,omitempty"`
+	CreatedAt time.Time `json:"createdAt"`
+}
+
+// Store persists todos. List returns newest first; an empty project means all.
+type Store interface {
+	Add(ctx context.Context, t Todo) error
+	List(ctx context.Context, project string, limit int) ([]Todo, error)
+}
+
+// FileStore keeps todos in memory and appends them to a JSONL file.
+type FileStore struct {
+	mu    sync.Mutex
+	path  string
+	items []Todo
+}
+
+func NewFileStore(path string) (*FileStore, error) {
+	fs := &FileStore{path: path}
+	f, err := os.Open(path)
+	if os.IsNotExist(err) {
+		return fs, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+	defer f.Close()
+	sc := bufio.NewScanner(f)
+	sc.Buffer(make([]byte, 64<<10), 1<<20)
+	for sc.Scan() {
+		var t Todo
+		if json.Unmarshal(sc.Bytes(), &t) == nil {
+			fs.items = append(fs.items, t)
+		}
+	}
+	return fs, sc.Err()
+}
+
+func (s *FileStore) Add(_ context.Context, t Todo) error {
+	line, err := json.Marshal(t)
+	if err != nil {
+		return err
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	f, err := os.OpenFile(s.path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+	if _, err := f.Write(append(line, '\n')); err != nil {
+		return err
+	}
+	s.items = append(s.items, t)
+	return nil
+}
+
+func (s *FileStore) List(_ context.Context, project string, limit int) ([]Todo, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	var out []Todo
+	for _, t := range s.items {
+		if project == "" || t.Project == project {
+			out = append(out, t)
+		}
+	}
+	sort.SliceStable(out, func(i, j int) bool { return out[i].CreatedAt.After(out[j].CreatedAt) })
+	if limit > 0 && len(out) > limit {
+		out = out[:limit]
+	}
+	return out, nil
+}
+
+// ClickHouseStore persists todos in a plain MergeTree table (no TTL).
+type ClickHouseStore struct {
+	conn driver.Conn
+	db   string
+}
+
+func NewClickHouseStore(ctx context.Context, conn driver.Conn, db string) (*ClickHouseStore, error) {
+	err := conn.Exec(ctx, fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s.todos (
+		id String,
+		project LowCardinality(String),
+		text String,
+		component String,
+		path String,
+		hash String,
+		url String,
+		user_agent String,
+		session String,
+		created_at DateTime64(3, 'UTC')
+	) ENGINE = MergeTree
+	ORDER BY (project, created_at)`, db))
+	if err != nil {
+		return nil, err
+	}
+	return &ClickHouseStore{conn: conn, db: db}, nil
+}
+
+func (s *ClickHouseStore) Add(ctx context.Context, t Todo) error {
+	return s.conn.Exec(ctx, fmt.Sprintf(`INSERT INTO %s.todos (id, project, text, component, path, hash, url, user_agent, session, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, s.db),
+		t.ID, t.Project, t.Text, t.Component, t.Path, t.Hash, t.URL, t.UserAgent, t.Session, t.CreatedAt.UTC())
+}
+
+func (s *ClickHouseStore) List(ctx context.Context, project string, limit int) ([]Todo, error) {
+	q := fmt.Sprintf(`SELECT id, project, text, component, path, hash, url, user_agent, session, created_at FROM %s.todos`, s.db)
+	var args []any
+	if project != "" {
+		q += ` WHERE project = ?`
+		args = append(args, project)
+	}
+	q += ` ORDER BY created_at DESC LIMIT ?`
+	args = append(args, limit)
+	rows, err := s.conn.Query(ctx, q, args...)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var out []Todo
+	for rows.Next() {
+		var t Todo
+		if err := rows.Scan(&t.ID, &t.Project, &t.Text, &t.Component, &t.Path, &t.Hash, &t.URL, &t.UserAgent, &t.Session, &t.CreatedAt); err != nil {
+			return nil, err
+		}
+		out = append(out, t)
+	}
+	return out, rows.Err()
+}