feat(server): revision authors, revisions after a seq, online machines

backend/internal/server/server.go

@@ -50,12 +50,14 @@ type Server struct {
 
 	mu          sync.Mutex
 	subscribers map[chan protocol.Event]struct{}
+	streams     map[int64]int
+	seen        map[int64]int64
 }
 
 type machineKey struct{}
 
 func New(store *Store, config Config) http.Handler {
-	s := &Server{config: config, store: store, subscribers: map[chan protocol.Event]struct{}{}}
+	s := &Server{config: config, store: store, subscribers: map[chan protocol.Event]struct{}{}, streams: map[int64]int{}, seen: map[int64]int64{}}
 	mux := http.NewServeMux()
 	mux.HandleFunc("GET /i/{code}", s.install)
 	mux.Handle("GET /dl/", http.StripPrefix("/dl/", http.FileServer(http.Dir(config.ClientsDir))))
@@ -136,6 +138,7 @@ func (s *Server) auth(next http.HandlerFunc) http.HandlerFunc {
 				return
 			}
 			machineID = id
+			s.track(id, 0)
 		}
 		if key := r.PathValue("key"); key != "" && !repoKeyPattern.MatchString(key) {
 			http.Error(w, "bad repository key", http.StatusBadRequest)
@@ -206,12 +209,25 @@ func (s *Server) logout(w http.ResponseWriter, r *http.Request) {
 	w.WriteHeader(http.StatusNoContent)
 }
 
+func (s *Server) track(machineID int64, streams int) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.seen[machineID] = time.Now().UnixMilli()
+	s.streams[machineID] += streams
+}
+
 func (s *Server) machines(w http.ResponseWriter, r *http.Request) {
 	machines, err := s.store.Machines(r.Context())
 	if err != nil {
 		writeError(w, err)
 		return
 	}
+	s.mu.Lock()
+	for i, machine := range machines {
+		machines[i].Online = s.streams[machine.ID] > 0
+		machines[i].LastSeen = max(s.seen[machine.ID], machine.LastPush)
+	}
+	s.mu.Unlock()
 	writeJSON(w, http.StatusOK, machines)
 }
 
@@ -287,7 +303,19 @@ func (s *Server) state(w http.ResponseWriter, r *http.Request) {
 }
 
 func (s *Server) revisions(w http.ResponseWriter, r *http.Request) {
-	revisions, err := s.store.Revisions(r.Context(), r.PathValue("key"), r.URL.Query().Get("path"))
+	key, query := r.PathValue("key"), r.URL.Query()
+	var revisions []protocol.Revision
+	var err error
+	if query.Has("after") {
+		after, parseErr := strconv.ParseInt(query.Get("after"), 10, 64)
+		if parseErr != nil {
+			http.Error(w, "bad after", http.StatusBadRequest)
+			return
+		}
+		revisions, err = s.store.RevisionsAfter(r.Context(), key, query.Get("path"), after)
+	} else {
+		revisions, err = s.store.Revisions(r.Context(), key, query.Get("path"))
+	}
 	if err != nil {
 		writeError(w, err)
 		return
@@ -319,8 +347,9 @@ func (s *Server) push(w http.ResponseWriter, r *http.Request) {
 		writeError(w, err)
 		return
 	}
-	s.publish(protocol.Event{Repo: key, Seq: seq})
-	writeJSON(w, http.StatusCreated, protocol.Event{Repo: key, Seq: seq})
+	event := protocol.Event{Repo: key, Seq: seq, Author: push.Author}
+	s.publish(event)
+	writeJSON(w, http.StatusCreated, event)
 }
 
 func (s *Server) blob(w http.ResponseWriter, r *http.Request) {
@@ -370,14 +399,17 @@ func (s *Server) events(w http.ResponseWriter, r *http.Request) {
 		http.Error(w, "streaming unsupported", http.StatusInternalServerError)
 		return
 	}
+	machineID, _ := r.Context().Value(machineKey{}).(int64)
 	subscriber := make(chan protocol.Event, 16)
 	s.mu.Lock()
 	s.subscribers[subscriber] = struct{}{}
 	s.mu.Unlock()
+	s.track(machineID, 1)
 	defer func() {
 		s.mu.Lock()
 		delete(s.subscribers, subscriber)
 		s.mu.Unlock()
+		s.track(machineID, -1)
 	}()
 
 	w.Header().Set("Content-Type", "text/event-stream")
@@ -391,6 +423,7 @@ func (s *Server) events(w http.ResponseWriter, r *http.Request) {
 		case <-r.Context().Done():
 			return
 		case <-ticker.C:
+			s.track(machineID, 0)
 			fmt.Fprint(w, ": ping\n\n")
 		case event := <-subscriber:
 			data, _ := json.Marshal(event)

backend/internal/server/server_test.go

@@ -0,0 +1,206 @@
+package server
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"crypto/sha256"
+	"database/sql"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"lodge/internal/protocol"
+)
+
+const (
+	adminToken = "admin"
+	repoKey    = "0123456789abcdef0123456789abcdef01234567"
+)
+
+func newServer(t *testing.T) string {
+	t.Helper()
+	store, err := OpenStore(filepath.Join(t.TempDir(), "lodge.db"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { store.Close() })
+	api := httptest.NewServer(New(store, Config{AdminToken: adminToken}))
+	t.Cleanup(api.Close)
+	return api.URL
+}
+
+func call(t *testing.T, method, url, token string, in, out any) {
+	t.Helper()
+	var body bytes.Buffer
+	if in != nil {
+		_ = json.NewEncoder(&body).Encode(in)
+	}
+	req, _ := http.NewRequest(method, url, &body)
+	req.Header.Set("Authorization", "Bearer "+token)
+	res, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer res.Body.Close()
+	if res.StatusCode >= http.StatusBadRequest {
+		t.Fatalf("%s %s: %s", method, url, res.Status)
+	}
+	if out != nil {
+		if err := json.NewDecoder(res.Body).Decode(out); err != nil {
+			t.Fatal(err)
+		}
+	}
+}
+
+func push(t *testing.T, url, author, path, base, content string) string {
+	t.Helper()
+	sum := sha256.Sum256([]byte(content))
+	hash := hex.EncodeToString(sum[:])
+	req, _ := http.NewRequest(http.MethodPut, url+"/api/blobs/"+hash, strings.NewReader(content))
+	req.Header.Set("Authorization", "Bearer "+adminToken)
+	res, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	res.Body.Close()
+	change := protocol.Change{Space: protocol.SharedSpace, Path: path, Base: base, Hash: hash}
+	call(t, http.MethodPost, url+"/api/repos/"+repoKey+"/revisions", adminToken, protocol.Push{Name: "vault", Author: author, Changes: []protocol.Change{change}}, nil)
+	return hash
+}
+
+func revisions(t *testing.T, url, query string) []protocol.Revision {
+	t.Helper()
+	var revisions []protocol.Revision
+	call(t, http.MethodGet, url+"/api/repos/"+repoKey+"/revisions"+query, adminToken, nil, &revisions)
+	return revisions
+}
+
+func seqs(revisions []protocol.Revision) []int64 {
+	seqs := make([]int64, len(revisions))
+	for i, rev := range revisions {
+		seqs[i] = rev.Seq
+	}
+	return seqs
+}
+
+func TestRevisionsCarryAuthorAndPageAfterSeq(t *testing.T) {
+	url := newServer(t)
+	events, err := http.NewRequest(http.MethodGet, url+"/api/events", nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	events.Header.Set("Authorization", "Bearer "+adminToken)
+	stream, err := http.DefaultClient.Do(events)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer stream.Body.Close()
+
+	first := push(t, url, "", "a.md", "", "one")
+	push(t, url, "alice", "a.md", first, "two")
+	push(t, url, "bob", "b.md", "", "three")
+
+	if got := seqs(revisions(t, url, "")); len(got) != 3 || got[0] != 3 {
+		t.Fatalf("default order = %v, want newest first", got)
+	}
+	after := revisions(t, url, "?after=1")
+	if got := seqs(after); len(got) != 2 || got[0] != 2 || got[1] != 3 {
+		t.Fatalf("after=1 = %v, want [2 3]", got)
+	}
+	if after[0].Author != "alice" || after[0].Machine != "web" || revisions(t, url, "?after=0")[0].Author != "" {
+		t.Fatalf("authors = %+v", after)
+	}
+	if got := seqs(revisions(t, url, "?after=0&path=b.md")); len(got) != 1 || got[0] != 3 {
+		t.Fatalf("after with path = %v", got)
+	}
+	if got := revisions(t, url, "?after=3"); len(got) != 0 {
+		t.Fatalf("after the last revision = %v", got)
+	}
+
+	var authors []string
+	scanner := bufio.NewScanner(stream.Body)
+	for len(authors) < 3 && scanner.Scan() {
+		if data, ok := strings.CutPrefix(scanner.Text(), "data: "); ok {
+			var event protocol.Event
+			_ = json.Unmarshal([]byte(data), &event)
+			authors = append(authors, event.Author)
+		}
+	}
+	if strings.Join(authors, ",") != ",alice,bob" {
+		t.Fatalf("event authors = %q", authors)
+	}
+}
+
+func TestOldDatabaseGainsAuthorColumn(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "lodge.db")
+	db, err := sql.Open("sqlite", path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	_, err = db.Exec(`
+		create table revisions (repo text not null, seq integer not null, machine_id integer, time integer not null, main_sha text not null, main_branch text not null, primary key (repo, seq));
+		create table changes (repo text not null, seq integer not null, space text not null, path text not null, base text not null, hash text not null, primary key (repo, seq, space, path));
+		insert into revisions values ('` + repoKey + `', 1, null, 1, '', 'main');
+		insert into changes values ('` + repoKey + `', 1, 'shared', 'a.md', '', 'x');`)
+	if err := errors.Join(err, db.Close()); err != nil {
+		t.Fatal(err)
+	}
+	for range 2 {
+		store, err := OpenStore(path)
+		if err != nil {
+			t.Fatal(err)
+		}
+		revisions, err := store.Revisions(context.Background(), repoKey, "")
+		store.Close()
+		if err != nil || len(revisions) != 1 || revisions[0].Author != "" {
+			t.Fatalf("old revisions = %+v (%v)", revisions, err)
+		}
+	}
+}
+
+func TestMachinesShowWhoIsOnline(t *testing.T) {
+	url := newServer(t)
+	var code protocol.CodeResponse
+	call(t, http.MethodPost, url+"/api/codes", adminToken, nil, &code)
+	var claim protocol.ClaimResponse
+	call(t, http.MethodPost, url+"/api/claim", "", protocol.ClaimRequest{Code: code.Code, Name: "dell"}, &claim)
+	online := func() protocol.Machine {
+		t.Helper()
+		var machines []protocol.Machine
+		call(t, http.MethodGet, url+"/api/machines", adminToken, nil, &machines)
+		if len(machines) != 1 {
+			t.Fatalf("machines = %+v", machines)
+		}
+		return machines[0]
+	}
+	if machine := online(); machine.Online || machine.LastSeen != 0 {
+		t.Fatalf("a machine that never connected: %+v", machine)
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	req, _ := http.NewRequestWithContext(ctx, http.MethodGet, url+"/api/events", nil)
+	req.Header.Set("Authorization", "Bearer "+claim.Token)
+	stream, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if machine := online(); !machine.Online || machine.LastSeen == 0 {
+		t.Fatalf("a machine listening for events: %+v", machine)
+	}
+	cancel()
+	stream.Body.Close()
+	deadline := time.Now().Add(5 * time.Second)
+	for online().Online {
+		if time.Now().After(deadline) {
+			t.Fatal("the machine stays online after its stream closed")
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+}

backend/internal/server/store.go

@@ -20,7 +20,12 @@ const (
 	codeLifetime    = 10 * time.Minute
 	sessionLifetime = 30 * 24 * time.Hour
 	revisionLimit   = 500
-	schema          = `
+	revisionRows    = `
+		select r.seq, coalesce(m.name, iif(r.machine_id is null, 'web', 'revoked')), r.author, r.time, r.main_sha, r.main_branch, c.space, c.path, c.base, c.hash
+		from revisions r
+		join changes c on c.repo = r.repo and c.seq = r.seq
+		left join machines m on m.id = r.machine_id`
+	schema = `
 create table if not exists machines (
 	id integer primary key,
 	name text not null,
@@ -49,6 +54,7 @@ create table if not exists revisions (
 	repo text not null,
 	seq integer not null,
 	machine_id integer,
+	author text not null default '',
 	time integer not null,
 	main_sha text not null,
 	main_branch text not null,
@@ -86,9 +92,21 @@ func OpenStore(path string) (*Store, error) {
 	if _, err := db.Exec(schema); err != nil {
 		return nil, err
 	}
+	if err := addAuthor(db); err != nil {
+		return nil, err
+	}
 	return &Store{db: db}, nil
 }
 
+func addAuthor(db *sql.DB) error {
+	var found int
+	if err := db.QueryRow(`select count(*) from pragma_table_info('revisions') where name = 'author'`).Scan(&found); err != nil || found > 0 {
+		return err
+	}
+	_, err := db.Exec(`alter table revisions add column author text not null default ''`)
+	return err
+}
+
 func (s *Store) Close() error {
 	return s.db.Close()
 }
@@ -292,14 +310,24 @@ func (s *Store) State(ctx context.Context, repo, at string) (protocol.State, err
 }
 
 func (s *Store) Revisions(ctx context.Context, repo, path string) ([]protocol.Revision, error) {
-	rows, err := s.db.QueryContext(ctx, `
-		select r.seq, coalesce(m.name, iif(r.machine_id is null, 'web', 'revoked')), r.time, r.main_sha, r.main_branch, c.space, c.path, c.base, c.hash
-		from revisions r
-		join changes c on c.repo = r.repo and c.seq = r.seq
-		left join machines m on m.id = r.machine_id
+	return s.revisions(ctx, revisionRows+`
 		where r.repo = ?1 and (?2 = '' or c.path = ?2)
 		order by r.seq desc, c.path
 		limit ?3`, repo, path, revisionLimit)
+}
+
+func (s *Store) RevisionsAfter(ctx context.Context, repo, path string, after int64) ([]protocol.Revision, error) {
+	return s.revisions(ctx, revisionRows+`
+		where r.repo = ?1 and (?2 = '' or c.path = ?2) and r.seq in (
+			select distinct seq from changes
+			where repo = ?1 and (?2 = '' or path = ?2) and seq > ?4
+			order by seq limit ?3
+		)
+		order by r.seq, c.path`, repo, path, revisionLimit, after)
+}
+
+func (s *Store) revisions(ctx context.Context, query string, args ...any) ([]protocol.Revision, error) {
+	rows, err := s.db.QueryContext(ctx, query, args...)
 	if err != nil {
 		return nil, err
 	}
@@ -308,7 +336,7 @@ func (s *Store) Revisions(ctx context.Context, repo, path string) ([]protocol.Re
 	for rows.Next() {
 		var rev protocol.Revision
 		var change protocol.Change
-		if err := rows.Scan(&rev.Seq, &rev.Machine, &rev.Time, &rev.MainSHA, &rev.MainBranch, &change.Space, &change.Path, &change.Base, &change.Hash); err != nil {
+		if err := rows.Scan(&rev.Seq, &rev.Machine, &rev.Author, &rev.Time, &rev.MainSHA, &rev.MainBranch, &change.Space, &change.Path, &change.Base, &change.Hash); err != nil {
 			return nil, err
 		}
 		if last := len(revisions) - 1; last >= 0 && revisions[last].Seq == rev.Seq {
@@ -356,7 +384,7 @@ func (s *Store) Push(ctx context.Context, repo string, machineID int64, push pro
 		return 0, err
 	}
 	machine := sql.NullInt64{Int64: machineID, Valid: machineID != 0}
-	if _, err := tx.ExecContext(ctx, `insert into revisions (repo, seq, machine_id, time, main_sha, main_branch) values (?, ?, ?, ?, ?, ?)`, repo, seq, machine, now, push.MainSHA, push.MainBranch); err != nil {
+	if _, err := tx.ExecContext(ctx, `insert into revisions (repo, seq, machine_id, author, time, main_sha, main_branch) values (?, ?, ?, ?, ?, ?, ?)`, repo, seq, machine, push.Author, now, push.MainSHA, push.MainBranch); err != nil {
 		return 0, err
 	}
 	for _, change := range push.Changes {