From c2e9284a630b0dae724cf2fb70d1941d351dbf31 Mon Sep 17 00:00:00 2001 From: h Date: Fri, 2 Oct 2026 22:07:08 +0000 Subject: [PATCH] feat(server): revision authors, revisions after a seq, online machines --- backend/cmd/lodge/main.go | 6 +- backend/internal/client/remote_test.go | 8 + backend/internal/protocol/protocol.go | 9 +- backend/internal/server/server.go | 41 ++++- backend/internal/server/server_test.go | 206 +++++++++++++++++++++++++ backend/internal/server/store.go | 44 +++++- 6 files changed, 299 insertions(+), 15 deletions(-) create mode 100644 backend/internal/server/server_test.go diff --git a/backend/cmd/lodge/main.go b/backend/cmd/lodge/main.go index c337308..7d4d305 100644 --- a/backend/cmd/lodge/main.go +++ b/backend/cmd/lodge/main.go @@ -265,7 +265,11 @@ func showLog(ctx context.Context, cfg *client.Config, path string, asJSON bool) return output(asJSON, revisions, func() { for _, rev := range revisions { when := time.UnixMilli(rev.Time).Format("2006-01-02 15:04") - fmt.Printf("%d %s %s %s@%.10s\n", rev.Seq, when, rev.Machine, rev.MainBranch, rev.MainSHA) + who := rev.Machine + if rev.Author != "" { + who = rev.Author + " via " + rev.Machine + } + fmt.Printf("%d %s %s %s@%.10s\n", rev.Seq, when, who, rev.MainBranch, rev.MainSHA) for _, change := range rev.Changes { action := "M" if change.Base == "" { diff --git a/backend/internal/client/remote_test.go b/backend/internal/client/remote_test.go index c8b064a..8e2f53b 100644 --- a/backend/internal/client/remote_test.go +++ b/backend/internal/client/remote_test.go @@ -193,6 +193,14 @@ func TestDaemonFollowsServersAddedLater(t *testing.T) { write(t, mac.root, "docs/agents/second.md", "second") mac.sync() fresh.waitFor("docs/agents/second.md", "second") + + var machines []protocol.Machine + if err := (&Client{URL: event, Token: adminToken}).call(context.Background(), http.MethodGet, "/api/machines", nil, &machines); err != nil { + t.Fatal(err) + } + if len(machines) != 1 || !machines[0].Online { + t.Fatalf("the daemon is not online on the server it was bound to: %+v", machines) + } } func (m *machine) waitFor(path, content string) { diff --git a/backend/internal/protocol/protocol.go b/backend/internal/protocol/protocol.go index e4a0113..0181d68 100644 --- a/backend/internal/protocol/protocol.go +++ b/backend/internal/protocol/protocol.go @@ -32,6 +32,7 @@ type Change struct { type Push struct { Name string `json:"name"` + Author string `json:"author,omitempty"` MainSHA string `json:"main_sha"` MainBranch string `json:"main_branch"` Changes []Change `json:"changes"` @@ -40,6 +41,7 @@ type Push struct { type Revision struct { Seq int64 `json:"seq"` Machine string `json:"machine"` + Author string `json:"author"` Time int64 `json:"time"` MainSHA string `json:"main_sha"` MainBranch string `json:"main_branch"` @@ -58,6 +60,8 @@ type Machine struct { Name string `json:"name"` CreatedAt int64 `json:"created_at"` LastPush int64 `json:"last_push"` + LastSeen int64 `json:"last_seen"` + Online bool `json:"online"` Revisions int64 `json:"revisions"` } @@ -66,8 +70,9 @@ type Login struct { } type Event struct { - Repo string `json:"repo"` - Seq int64 `json:"seq"` + Repo string `json:"repo"` + Seq int64 `json:"seq"` + Author string `json:"author,omitempty"` } type ClaimRequest struct { diff --git a/backend/internal/server/server.go b/backend/internal/server/server.go index 51c9bbf..b966279 100644 --- a/backend/internal/server/server.go +++ b/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) diff --git a/backend/internal/server/server_test.go b/backend/internal/server/server_test.go new file mode 100644 index 0000000..d893414 --- /dev/null +++ b/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) + } +} diff --git a/backend/internal/server/store.go b/backend/internal/server/store.go index f27f221..8429aa1 100644 --- a/backend/internal/server/store.go +++ b/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 {