ci

 1@@ -12,3 +12,12 @@ linters:
 2         - (*os.File).Write
 3         - (io.Closer).Close
 4         - (*git.kilimanjaro.io/ci/internal/runlog.Writer).Close
 5+        - (*git.kilimanjaro.io/ci/internal/eventlog.Log).Close
 6+        - (*database/sql.DB).Close
 7+        - (*database/sql.Rows).Close
 8+        - (*github.com/gorilla/websocket.Conn).Close
 9+  exclusions:
10+    rules:
11+      # Test setup errors surface as test failures anyway.
12+      - path: _test\.go
13+        linters: [errcheck]

 1@@ -138,16 +138,28 @@ split across hosts later.
 2 ## 5. Bridge (forge side)
 3 
 4 - A line appended to the existing soft-serve `post-receive` hook (which
 5-  already rebuilds the pgit site) calls `ci bridge notify <repo> <old> <new>
 6-  <ref>`. It must never block or fail a push.
 7+  already rebuilds the pgit site) runs `ci bridge notify --spool <dir>
 8+  --server <bridge>`. It reads git's `<old> <new> <ref>` lines on stdin and
 9+  soft-serve's `SOFT_SERVE_REPO_NAME`/`SOFT_SERVE_REPO_PATH`, computes each
10+  update from the bare repo (changed files, commits per author, default
11+  ref, push options), writes one JSON file per push into the spool, pokes
12+  the bridge, and always exits 0: it never blocks or fails a push.
13+- The bridge server ingests the spool on start, on every poke and every
14+  10 s, and deletes each file once its events are in the log; unparsable
15+  files are moved aside (`.bad`). A bridge that is down only delays events.
16+  Repos are `did:web:<forge-host>:<repo>`; `ownerDid` and `committerDid` are
17+  the configured owner DID (soft-serve users have no DIDs).
18 - For each pushed ref the bridge emits one `sh.tangled.git.refUpdate` event:
19   repo DID, ref, old/new sha, changed files, push options (`-o skip-ci` is
20   honoured by the runner, as upstream). The bridge does not read workflows;
21   that moved to the spindle upstream, and the runner does the same (§10).
22 - Manual triggers (appview "Run", re-run, `ci trigger`) go to the runner's
23   `sh.tangled.ci.triggerPipeline`, as upstream.
24-- Event log: SQLite `(seq INTEGER PRIMARY KEY, nsid, rkey, json,
25-  created_at)`. Reconnect and crash recovery are "resume from cursor".
26+- Event log (`internal/eventlog`, shared with the runner): SQLite
27+  `(created INTEGER PRIMARY KEY, rkey, nsid, event)`, where `created` is a
28+  monotonic UnixNano and the cursor, as upstream's `eventstream`.
29+  Reconnect and crash recovery are "resume from cursor". `/events` needs the
30+  bearer token when one is configured.
31 - Routing: one runner, configured in the bridge. (Tangled's model — the repo
32   names its spindle — is the extension point if a second runner ever exists.)
33 

  1@@ -0,0 +1,157 @@
  2+package main
  3+
  4+import (
  5+	"context"
  6+	"errors"
  7+	"flag"
  8+	"fmt"
  9+	"log/slog"
 10+	"net/http"
 11+	"os"
 12+	"os/signal"
 13+	"strings"
 14+	"syscall"
 15+	"time"
 16+
 17+	"git.kilimanjaro.io/ci/internal/bridge"
 18+	"git.kilimanjaro.io/ci/internal/eventlog"
 19+)
 20+
 21+func init() {
 22+	commands["bridge"] = command{"forge bridge: knot-compatible event stream (serve | notify)", cmdBridge}
 23+}
 24+
 25+func cmdBridge(args []string) int {
 26+	if len(args) == 0 {
 27+		fmt.Fprintln(os.Stderr, "usage: ci bridge serve [flags] | ci bridge notify [flags]")
 28+		return exitUsage
 29+	}
 30+	switch args[0] {
 31+	case "serve":
 32+		return bridgeServe(args[1:])
 33+	case "notify":
 34+		return bridgeNotify(args[1:])
 35+	}
 36+	fmt.Fprintf(os.Stderr, "ci bridge: unknown subcommand %q\n", args[0])
 37+	return exitUsage
 38+}
 39+
 40+// readToken reads a token file (empty path: no token).
 41+func readToken(path string) (string, error) {
 42+	if path == "" {
 43+		return "", nil
 44+	}
 45+	b, err := os.ReadFile(path)
 46+	if err != nil {
 47+		return "", err
 48+	}
 49+	t := strings.TrimSpace(string(b))
 50+	if t == "" {
 51+		return "", fmt.Errorf("%s is empty", path)
 52+	}
 53+	return t, nil
 54+}
 55+
 56+func bridgeServe(args []string) int {
 57+	fs := flag.NewFlagSet("ci bridge serve", flag.ContinueOnError)
 58+	listen := fs.String("listen", "127.0.0.1:8475", "address to serve on")
 59+	db := fs.String("db", "bridge.db", "event log (SQLite)")
 60+	spool := fs.String("spool", "spool", "spool directory written by `ci bridge notify`")
 61+	host := fs.String("forge-host", "", "forge host name: repos are did:web:<host>:<repo> (required)")
 62+	owner := fs.String("owner-did", "", "DID of the repos' owner (default did:web:<forge-host>)")
 63+	tokenFile := fs.String("token-file", "", "require this bearer token on /events and /ingest")
 64+	if err := fs.Parse(args); err != nil {
 65+		return exitUsage
 66+	}
 67+	if *host == "" {
 68+		fmt.Fprintln(os.Stderr, "ci bridge serve: --forge-host is required")
 69+		return exitUsage
 70+	}
 71+	token, err := readToken(*tokenFile)
 72+	if err != nil {
 73+		fmt.Fprintln(os.Stderr, "ci bridge serve:", err)
 74+		return exitFailure
 75+	}
 76+	log, err := eventlog.Open(*db)
 77+	if err != nil {
 78+		fmt.Fprintln(os.Stderr, "ci bridge serve:", err)
 79+		return exitFailure
 80+	}
 81+	defer log.Close()
 82+	s := &bridge.Server{Log: log, SpoolDir: *spool, ForgeHost: *host, OwnerDid: *owner, Token: token, Logger: slog.Default()}
 83+	return serve(*listen, s.Handler(), func(ctx context.Context) { s.Run(ctx, 10*time.Second) })
 84+}
 85+
 86+// serve runs an HTTP server and a background loop until SIGINT/SIGTERM.
 87+func serve(listen string, h http.Handler, loop func(context.Context)) int {
 88+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
 89+	defer stop()
 90+	srv := &http.Server{Addr: listen, Handler: h, ReadHeaderTimeout: 10 * time.Second}
 91+	if loop != nil {
 92+		go loop(ctx)
 93+	}
 94+	go func() {
 95+		<-ctx.Done()
 96+		sctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 97+		defer cancel()
 98+		_ = srv.Shutdown(sctx)
 99+	}()
100+	slog.Info("listening", "addr", listen)
101+	if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
102+		fmt.Fprintln(os.Stderr, "ci:", err)
103+		return exitFailure
104+	}
105+	return exitSuccess
106+}
107+
108+// bridgeNotify is the post-receive hook. It always exits 0: a broken bridge
109+// must never fail a push.
110+func bridgeNotify(args []string) int {
111+	fs := flag.NewFlagSet("ci bridge notify", flag.ContinueOnError)
112+	spool := fs.String("spool", "", "spool directory (required)")
113+	server := fs.String("server", "", "bridge URL to poke after spooling (best effort)")
114+	tokenFile := fs.String("token-file", "", "bearer token for the poke")
115+	gitDir := fs.String("git-dir", "", "bare repository (default $SOFT_SERVE_REPO_PATH, else $GIT_DIR, else .)")
116+	repo := fs.String("repo", "", "repository name (default $SOFT_SERVE_REPO_NAME)")
117+	if err := fs.Parse(args); err != nil {
118+		return exitSuccess
119+	}
120+	warn := func(err error) int {
121+		fmt.Fprintln(os.Stderr, "ci bridge notify:", err)
122+		return exitSuccess
123+	}
124+	if *spool == "" {
125+		return warn(errors.New("--spool is required"))
126+	}
127+	name := firstNonEmpty(*repo, os.Getenv("SOFT_SERVE_REPO_NAME"))
128+	if name == "" {
129+		return warn(errors.New("no repository name (--repo or $SOFT_SERVE_REPO_NAME)"))
130+	}
131+	dir := firstNonEmpty(*gitDir, os.Getenv("SOFT_SERVE_REPO_PATH"), os.Getenv("GIT_DIR"), ".")
132+	lines, err := bridge.ParseHookInput(os.Stdin)
133+	if err != nil {
134+		return warn(err)
135+	}
136+	ctx := context.Background()
137+	p, err := bridge.BuildPush(ctx, dir, strings.TrimSuffix(name, ".git"), os.Getenv("SOFT_SERVE_USERNAME"), lines, bridge.PushOptions(os.Getenv))
138+	if err != nil {
139+		return warn(err)
140+	}
141+	token, err := readToken(*tokenFile)
142+	if err != nil {
143+		return warn(err)
144+	}
145+	if err := bridge.Notify(ctx, *spool, *server, token, p); err != nil {
146+		return warn(err)
147+	}
148+	return exitSuccess
149+}
150+
151+func firstNonEmpty(s ...string) string {
152+	for _, x := range s {
153+		if x != "" {
154+			return x
155+		}
156+	}
157+	return ""
158+}

1@@ -34,7 +34,7 @@
2               ./lexicons
3             ];
4           };
5-          vendorHash = "sha256-xWBzri5Qza/1abQHZtOhnRzRjw57lOrCZOhIQqoftOE=";
6+          vendorHash = "sha256-2lb+pmVbboWhcg5vFHPGaP9imrXvFGis+I+eWEIcu0g=";
7           subPackages = [ "cmd/ci" ];
8           env.CGO_ENABLED = "0";
9           ldflags = [ "-X main.version=${version}" ];

 1@@ -1,8 +1,19 @@
 2 module git.kilimanjaro.io/ci
 3 
 4-go 1.26
 5+go 1.26.0
 6 
 7 require (
 8 	github.com/bmatcuk/doublestar/v4 v4.10.3 // indirect
 9+	github.com/dustin/go-humanize v1.0.1 // indirect
10+	github.com/google/uuid v1.6.0 // indirect
11+	github.com/gorilla/websocket v1.5.3 // indirect
12+	github.com/mattn/go-isatty v0.0.24 // indirect
13+	github.com/ncruces/go-strftime v1.0.0 // indirect
14+	github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
15+	golang.org/x/sys v0.48.0 // indirect
16 	gopkg.in/yaml.v3 v3.0.1 // indirect
17+	modernc.org/libc v1.77.1 // indirect
18+	modernc.org/mathutil v1.7.1 // indirect
19+	modernc.org/memory v1.12.1 // indirect
20+	modernc.org/sqlite v1.60.1 // indirect
21 )

 1@@ -1,5 +1,27 @@
 2 github.com/bmatcuk/doublestar/v4 v4.10.3 h1:1UIv1+fqH3++ViVkpnaPTB+Pi1UHycnPFAq14zgK+gI=
 3 github.com/bmatcuk/doublestar/v4 v4.10.3/go.mod h1:xBQ8jztBU6kakFMg+8WGxn0c6z1fTSPVIjEY1Wr7jzc=
 4+github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY=
 5+github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
 6+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
 7+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 8+github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
 9+github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
10+github.com/mattn/go-isatty v0.0.24 h1:tGZZoVgT/KiqK1c8ocVLeDS8BSWMRd47J3Lbz7vsReI=
11+github.com/mattn/go-isatty v0.0.24/go.mod h1:nMCL3Zebbrt45jsMDgnfIwz6ydEQApk5oEI3HqDio6A=
12+github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOFAw7w=
13+github.com/ncruces/go-strftime v1.0.0/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls=
14+github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE=
15+github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
16+golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo=
17+golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
18 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
19 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
20 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
21+modernc.org/libc v1.77.1 h1:Ct8j47QtiZ1Enj2DtFXQtUqrPCAjdCmPjtCuvrYQ0Hs=
22+modernc.org/libc v1.77.1/go.mod h1:87/pZ4L6nD1zqW4nItuS12YO7hN1igAah34xjnQo/W0=
23+modernc.org/mathutil v1.7.1 h1:GCZVGXdaN8gTqB1Mf/usp1Y/hSqgI2vAGGP4jZMCxOU=
24+modernc.org/mathutil v1.7.1/go.mod h1:4p5IwJITfppl0G4sUEDtCr4DthTaT47/N3aT6MhfgJg=
25+modernc.org/memory v1.12.1 h1:nFMiWrpStgZczNl6XI9GnIk/rWhYIyHGUaR04pGbp9g=
26+modernc.org/memory v1.12.1/go.mod h1:/JP4VbVC+K5sU2wZi9bHoq2MAkCnrt2r98UGeSK7Mjw=
27+modernc.org/sqlite v1.60.1 h1:/blz53O951KWFOso4QQvEs/Fq6cDBKLtMVrYNSeJVKw=
28+modernc.org/sqlite v1.60.1/go.mod h1:1dIoEagfDE72QytD5scH1lxARtaUgKgHC/NuApA27r0=

  1@@ -0,0 +1,148 @@
  2+// Package bridge makes a soft-serve forge look like a Tangled knot: a
  3+// post-receive hook spools every push, and the bridge server turns spooled
  4+// pushes into sh.tangled.git.refUpdate events on a knot-compatible /events
  5+// stream (DESIGN.md §5). It reads no workflows; the runner does.
  6+//
  7+// The hook (ci bridge notify) runs as soft-serve with the bare repository at
  8+// hand, so it computes everything from git there and writes one JSON file
  9+// per push into the spool directory. That never blocks or fails a push, and
 10+// a bridge that is down only delays events: the server ingests the spool when
 11+// it starts, on every poke from a hook, and every few seconds.
 12+package bridge
 13+
 14+import (
 15+	"bufio"
 16+	"context"
 17+	"encoding/json"
 18+	"fmt"
 19+	"io"
 20+	"os"
 21+	"os/exec"
 22+	"path/filepath"
 23+	"sort"
 24+	"strconv"
 25+	"strings"
 26+
 27+	"git.kilimanjaro.io/ci/internal/lex"
 28+	"git.kilimanjaro.io/ci/internal/tid"
 29+)
 30+
 31+const zeroSha = "0000000000000000000000000000000000000000"
 32+
 33+// maxChangedFiles caps changedFiles; beyond it the list is left out, which
 34+// upstream treats as unknown (a paths filter then matches).
 35+const maxChangedFiles = 5000
 36+
 37+// Push is one spooled push: what the hook knew. DIDs are added by the server.
 38+type Push struct {
 39+	Repo     string             `json:"repo"`
 40+	Username string             `json:"username,omitempty"`
 41+	Updates  []lex.GitRefUpdate `json:"updates"`
 42+}
 43+
 44+// RefLine is one line of post-receive input: "<old> <new> <ref>".
 45+type RefLine struct{ Old, New, Ref string }
 46+
 47+// ParseHookInput reads post-receive's stdin.
 48+func ParseHookInput(r io.Reader) ([]RefLine, error) {
 49+	var out []RefLine
 50+	sc := bufio.NewScanner(r)
 51+	for sc.Scan() {
 52+		f := strings.Fields(sc.Text())
 53+		if len(f) == 0 {
 54+			continue
 55+		}
 56+		if len(f) != 3 {
 57+			return nil, fmt.Errorf("unexpected hook input %q", sc.Text())
 58+		}
 59+		out = append(out, RefLine{f[0], f[1], f[2]})
 60+	}
 61+	return out, sc.Err()
 62+}
 63+
 64+// PushOptions returns git's push options from the hook environment
 65+// (GIT_PUSH_OPTION_COUNT, GIT_PUSH_OPTION_<i>).
 66+func PushOptions(getenv func(string) string) []string {
 67+	n, _ := strconv.Atoi(getenv("GIT_PUSH_OPTION_COUNT"))
 68+	var out []string
 69+	for i := range n {
 70+		out = append(out, getenv("GIT_PUSH_OPTION_"+strconv.Itoa(i)))
 71+	}
 72+	return out
 73+}
 74+
 75+// BuildPush computes the ref updates of a push from the bare repository at
 76+// gitDir (after the refs were updated, as in post-receive).
 77+func BuildPush(ctx context.Context, gitDir, repo, username string, lines []RefLine, pushOptions []string) (Push, error) {
 78+	p := Push{Repo: repo, Username: username}
 79+	head, _ := git(ctx, gitDir, "symbolic-ref", "-q", "HEAD")
 80+	head = strings.TrimSpace(head)
 81+	for _, l := range lines {
 82+		u := lex.GitRefUpdate{
 83+			LexiconTypeID: lex.GitRefUpdateNSID,
 84+			Ref:           l.Ref,
 85+			OldSha:        l.Old,
 86+			NewSha:        l.New,
 87+			PushOptions:   pushOptions,
 88+			Meta:          &lex.GitRefUpdateMeta{IsDefaultRef: l.Ref == head, CommitCount: &lex.GitRefUpdateCommitCount{}},
 89+		}
 90+		if l.New != zeroSha {
 91+			rng := l.Old + ".." + l.New
 92+			if l.Old == zeroSha {
 93+				// a new ref: count what isn't on the default branch yet
 94+				rng = l.New
 95+				if head != "" && head != l.Ref {
 96+					if _, err := git(ctx, gitDir, "rev-parse", "--verify", "-q", head); err == nil {
 97+						rng = l.New + " ^" + head
 98+					}
 99+				}
100+			} else if out, err := git(ctx, gitDir, "diff", "--name-only", "--no-renames", l.Old, l.New); err == nil {
101+				files := strings.Fields(out)
102+				if len(files) <= maxChangedFiles {
103+					u.ChangedFiles = files
104+				}
105+			}
106+			if out, err := git(ctx, gitDir, append([]string{"log", "--format=%ae"}, strings.Fields(rng)...)...); err == nil {
107+				u.Meta.CommitCount.ByEmail = countEmails(out)
108+			}
109+		}
110+		p.Updates = append(p.Updates, u)
111+	}
112+	return p, nil
113+}
114+
115+func countEmails(log string) []lex.GitRefUpdateEmailCommitCount {
116+	n := map[string]int64{}
117+	for _, e := range strings.Fields(log) {
118+		n[e]++
119+	}
120+	out := make([]lex.GitRefUpdateEmailCommitCount, 0, len(n))
121+	for e, c := range n {
122+		out = append(out, lex.GitRefUpdateEmailCommitCount{Email: e, Count: c})
123+	}
124+	sort.Slice(out, func(i, j int) bool { return out[i].Email < out[j].Email })
125+	return out
126+}
127+
128+// Spool writes p into dir atomically as <tid>.json (sorted by time).
129+func Spool(dir string, p Push) (string, error) {
130+	if err := os.MkdirAll(dir, 0o775); err != nil {
131+		return "", err
132+	}
133+	b, err := json.Marshal(p)
134+	if err != nil {
135+		return "", err
136+	}
137+	name := filepath.Join(dir, tid.New()+".json")
138+	tmp := name + ".tmp"
139+	if err := os.WriteFile(tmp, b, 0o664); err != nil {
140+		return "", err
141+	}
142+	return name, os.Rename(tmp, name)
143+}
144+
145+func git(ctx context.Context, dir string, args ...string) (string, error) {
146+	cmd := exec.CommandContext(ctx, "git", append([]string{"--git-dir", dir}, args...)...)
147+	out, err := cmd.Output()
148+	return string(out), err
149+}

  1@@ -0,0 +1,166 @@
  2+package bridge
  3+
  4+import (
  5+	"context"
  6+	"encoding/json"
  7+	"net/http"
  8+	"net/http/httptest"
  9+	"os"
 10+	"os/exec"
 11+	"path/filepath"
 12+	"strings"
 13+	"testing"
 14+
 15+	"git.kilimanjaro.io/ci/internal/eventlog"
 16+	"git.kilimanjaro.io/ci/internal/lex"
 17+)
 18+
 19+func sh(t *testing.T, dir string, args ...string) string {
 20+	t.Helper()
 21+	cmd := exec.Command(args[0], args[1:]...)
 22+	cmd.Dir = dir
 23+	cmd.Env = append(os.Environ(), "GIT_AUTHOR_NAME=a", "GIT_AUTHOR_EMAIL=a@x", "GIT_COMMITTER_NAME=a",
 24+		"GIT_COMMITTER_EMAIL=a@x", "GIT_CONFIG_GLOBAL=/dev/null", "GIT_CONFIG_SYSTEM=/dev/null")
 25+	out, err := cmd.CombinedOutput()
 26+	if err != nil {
 27+		t.Fatalf("%v: %v\n%s", args, err, out)
 28+	}
 29+	return strings.TrimSpace(string(out))
 30+}
 31+
 32+func write(t *testing.T, p, s string) {
 33+	t.Helper()
 34+	if err := os.MkdirAll(filepath.Dir(p), 0o755); err != nil {
 35+		t.Fatal(err)
 36+	}
 37+	if err := os.WriteFile(p, []byte(s), 0o644); err != nil {
 38+		t.Fatal(err)
 39+	}
 40+}
 41+
 42+// repo makes a bare repo with two commits on main and returns it and the shas.
 43+func repo(t *testing.T) (bare, c1, c2, work string) {
 44+	t.Helper()
 45+	work = filepath.Join(t.TempDir(), "w")
 46+	os.MkdirAll(work, 0o755)
 47+	sh(t, work, "git", "init", "-q", "-b", "main")
 48+	write(t, filepath.Join(work, "a.txt"), "1")
 49+	sh(t, work, "git", "add", "-A")
 50+	sh(t, work, "git", "commit", "-qm", "one")
 51+	c1 = sh(t, work, "git", "rev-parse", "HEAD")
 52+	write(t, filepath.Join(work, "docs/b.md"), "2")
 53+	write(t, filepath.Join(work, "a.txt"), "1b")
 54+	sh(t, work, "git", "add", "-A")
 55+	sh(t, work, "git", "commit", "-qm", "two")
 56+	c2 = sh(t, work, "git", "rev-parse", "HEAD")
 57+	bare = filepath.Join(t.TempDir(), "r.git")
 58+	sh(t, "", "git", "clone", "-q", "--bare", work, bare)
 59+	return
 60+}
 61+
 62+func TestBuildPush(t *testing.T) {
 63+	ctx := context.Background()
 64+	bare, c1, c2, _ := repo(t)
 65+	lines, err := ParseHookInput(strings.NewReader(c1 + " " + c2 + " refs/heads/main\n" + zeroSha + " " + c2 + " refs/heads/feature\n" + c1 + " " + zeroSha + " refs/heads/old\n"))
 66+	if err != nil {
 67+		t.Fatal(err)
 68+	}
 69+	p, err := BuildPush(ctx, bare, "r", "bt", lines, []string{"skip-ci"})
 70+	if err != nil {
 71+		t.Fatal(err)
 72+	}
 73+	if len(p.Updates) != 3 {
 74+		t.Fatalf("%+v", p)
 75+	}
 76+	main := p.Updates[0]
 77+	if !main.Meta.IsDefaultRef || strings.Join(main.ChangedFiles, ",") != "a.txt,docs/b.md" || main.PushOptions[0] != "skip-ci" {
 78+		t.Errorf("main: %+v %+v", main, main.Meta)
 79+	}
 80+	if c := main.Meta.CommitCount.ByEmail; len(c) != 1 || c[0].Email != "a@x" || c[0].Count != 1 {
 81+		t.Errorf("commit count %+v", c)
 82+	}
 83+	feature := p.Updates[1]
 84+	if feature.Meta.IsDefaultRef || feature.ChangedFiles != nil {
 85+		t.Errorf("new branch: changed files must be unknown: %+v", feature)
 86+	}
 87+	if c := feature.Meta.CommitCount.ByEmail; len(c) != 0 {
 88+		t.Errorf("new branch at main's tip has no new commits: %+v", c)
 89+	}
 90+	if del := p.Updates[2]; del.NewSha != zeroSha {
 91+		t.Errorf("delete: %+v", del)
 92+	}
 93+}
 94+
 95+func TestPushOptions(t *testing.T) {
 96+	env := map[string]string{"GIT_PUSH_OPTION_COUNT": "2", "GIT_PUSH_OPTION_0": "ci-skip", "GIT_PUSH_OPTION_1": "x=y"}
 97+	got := PushOptions(func(k string) string { return env[k] })
 98+	if strings.Join(got, ",") != "ci-skip,x=y" {
 99+		t.Error(got)
100+	}
101+	if PushOptions(func(string) string { return "" }) != nil {
102+		t.Error("no options")
103+	}
104+}
105+
106+func TestServerIngestAndServe(t *testing.T) {
107+	ctx := context.Background()
108+	dir := t.TempDir()
109+	log, err := eventlog.Open(filepath.Join(dir, "bridge.db"))
110+	if err != nil {
111+		t.Fatal(err)
112+	}
113+	defer log.Close()
114+	s := &Server{Log: log, SpoolDir: filepath.Join(dir, "spool"), ForgeHost: "git.example", OwnerDid: "did:plc:owner", Token: "t"}
115+	srv := httptest.NewServer(s.Handler())
116+	defer srv.Close()
117+
118+	bare, c1, c2, _ := repo(t)
119+	p, _ := BuildPush(ctx, bare, "sovrn", "bt", []RefLine{{c1, c2, "refs/heads/main"}}, nil)
120+	// the hook spools and pokes; the server isn't running Run here, so ingest directly
121+	if err := Notify(ctx, s.SpoolDir, srv.URL, "t", p); err != nil {
122+		t.Fatal(err)
123+	}
124+	os.WriteFile(filepath.Join(s.SpoolDir, "zz-garbage.json"), []byte("{"), 0o644)
125+	n, err := s.Ingest()
126+	if err != nil || n != 1 {
127+		t.Fatalf("ingested %d: %v", n, err)
128+	}
129+	if left, _ := filepath.Glob(filepath.Join(s.SpoolDir, "*.json")); len(left) != 0 {
130+		t.Errorf("spool not emptied: %v", left)
131+	}
132+	if bad, _ := filepath.Glob(filepath.Join(s.SpoolDir, "*.bad")); len(bad) != 1 {
133+		t.Errorf("garbage not moved aside: %v", bad)
134+	}
135+	evs, _ := log.List(0, 10)
136+	if len(evs) != 1 || evs[0].Nsid != lex.GitRefUpdateNSID {
137+		t.Fatalf("%+v", evs)
138+	}
139+	var u lex.GitRefUpdate
140+	json.Unmarshal(evs[0].Event, &u)
141+	if u.Repo != "did:web:git.example:sovrn" || u.CommitterDid != "did:plc:owner" || *u.OwnerDid != "did:plc:owner" || u.LexiconTypeID != lex.GitRefUpdateNSID {
142+		t.Errorf("stamped %+v", u)
143+	}
144+
145+	// the stream needs the token; the DID document and health don't
146+	for path, want := range map[string]int{"/events": 401, "/.well-known/did.json": 200, "/healthz": 200} {
147+		resp, err := http.Get(srv.URL + path)
148+		if err != nil {
149+			t.Fatal(err)
150+		}
151+		resp.Body.Close()
152+		if resp.StatusCode != want {
153+			t.Errorf("%s: %d, want %d", path, resp.StatusCode, want)
154+		}
155+	}
156+}
157+
158+func TestNotifyWithoutServerStillSpools(t *testing.T) {
159+	dir := t.TempDir()
160+	p := Push{Repo: "r", Updates: []lex.GitRefUpdate{{Ref: "refs/heads/main"}}}
161+	if err := Notify(context.Background(), dir, "http://127.0.0.1:1", "", p); err != nil {
162+		t.Fatal(err)
163+	}
164+	if files, _ := filepath.Glob(filepath.Join(dir, "*.json")); len(files) != 1 {
165+		t.Errorf("spooled %v", files)
166+	}
167+}

  1@@ -0,0 +1,186 @@
  2+package bridge
  3+
  4+import (
  5+	"context"
  6+	"encoding/json"
  7+	"fmt"
  8+	"log/slog"
  9+	"net/http"
 10+	"os"
 11+	"path/filepath"
 12+	"sort"
 13+	"strings"
 14+	"sync"
 15+	"time"
 16+
 17+	"git.kilimanjaro.io/ci/internal/eventlog"
 18+	"git.kilimanjaro.io/ci/internal/ident"
 19+	"git.kilimanjaro.io/ci/internal/lex"
 20+)
 21+
 22+// Server ingests the spool into the event log and serves the stream.
 23+type Server struct {
 24+	Log      *eventlog.Log
 25+	SpoolDir string
 26+	// ForgeHost names repos: did:web:<ForgeHost>:<repo>.
 27+	ForgeHost string
 28+	// OwnerDid is the owner (and, for lack of per-user DIDs, the committer)
 29+	// of every repo. Default: did:web:<ForgeHost>.
 30+	OwnerDid string
 31+	// Token, if set, is required as a bearer token on /events and /ingest.
 32+	Token  string
 33+	Logger *slog.Logger
 34+
 35+	mu   sync.Mutex
 36+	poke chan struct{}
 37+}
 38+
 39+func (s *Server) owner() string {
 40+	if s.OwnerDid != "" {
 41+		return s.OwnerDid
 42+	}
 43+	return ident.ServiceDID(s.ForgeHost)
 44+}
 45+
 46+func (s *Server) logger() *slog.Logger {
 47+	if s.Logger != nil {
 48+		return s.Logger
 49+	}
 50+	return slog.Default()
 51+}
 52+
 53+// Ingest appends the events of every spooled push, oldest first, and deletes
 54+// each file once its events are in the log. A file that can't be parsed is
 55+// moved aside (<name>.bad) so it can't block the spool.
 56+func (s *Server) Ingest() (int, error) {
 57+	s.mu.Lock()
 58+	defer s.mu.Unlock()
 59+	files, err := filepath.Glob(filepath.Join(s.SpoolDir, "*.json"))
 60+	if err != nil {
 61+		return 0, err
 62+	}
 63+	sort.Strings(files)
 64+	n := 0
 65+	for _, f := range files {
 66+		b, err := os.ReadFile(f)
 67+		if err != nil {
 68+			return n, err
 69+		}
 70+		var p Push
 71+		if err := json.Unmarshal(b, &p); err != nil || p.Repo == "" {
 72+			s.logger().Error("bad spool file", "file", f, "err", err)
 73+			_ = os.Rename(f, f+".bad")
 74+			continue
 75+		}
 76+		owner := s.owner()
 77+		for _, u := range p.Updates {
 78+			u.LexiconTypeID = lex.GitRefUpdateNSID
 79+			u.Repo = ident.RepoDID(s.ForgeHost, p.Repo)
 80+			u.OwnerDid = &owner
 81+			u.CommitterDid = owner
 82+			if u.Meta == nil {
 83+				u.Meta = &lex.GitRefUpdateMeta{CommitCount: &lex.GitRefUpdateCommitCount{}}
 84+			}
 85+			if _, err := s.Log.Append(lex.GitRefUpdateNSID, "", u); err != nil {
 86+				return n, err
 87+			}
 88+			n++
 89+			s.logger().Info("ref update", "repo", p.Repo, "ref", u.Ref, "new", u.NewSha)
 90+		}
 91+		if err := os.Remove(f); err != nil {
 92+			return n, err
 93+		}
 94+	}
 95+	return n, nil
 96+}
 97+
 98+// Run ingests on start, whenever /ingest is poked, and every interval.
 99+func (s *Server) Run(ctx context.Context, interval time.Duration) {
100+	s.mu.Lock()
101+	if s.poke == nil {
102+		s.poke = make(chan struct{}, 1)
103+	}
104+	poke := s.poke
105+	s.mu.Unlock()
106+	for {
107+		if _, err := s.Ingest(); err != nil {
108+			s.logger().Error("ingest", "err", err)
109+		}
110+		select {
111+		case <-ctx.Done():
112+			return
113+		case <-poke:
114+		case <-time.After(interval):
115+		}
116+	}
117+}
118+
119+// Handler serves:
120+//
121+//	GET  /events                 the knot-compatible event stream
122+//	POST /ingest                 a hook's poke: ingest the spool now
123+//	GET  /.well-known/did.json   the bridge's did:web document
124+//	GET  /healthz                200 when the log is readable
125+func (s *Server) Handler() http.Handler {
126+	mux := http.NewServeMux()
127+	mux.Handle("GET /events", eventlog.RequireToken(s.Token, s.Log))
128+	mux.Handle("POST /ingest", eventlog.RequireToken(s.Token, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
129+		s.mu.Lock()
130+		if s.poke == nil {
131+			s.poke = make(chan struct{}, 1)
132+		}
133+		select {
134+		case s.poke <- struct{}{}:
135+		default:
136+		}
137+		s.mu.Unlock()
138+		w.WriteHeader(http.StatusAccepted)
139+	})))
140+	mux.HandleFunc("GET /.well-known/did.json", func(w http.ResponseWriter, r *http.Request) {
141+		w.Header().Set("Content-Type", "application/did+json")
142+		_ = json.NewEncoder(w).Encode(map[string]any{
143+			"@context": []string{"https://www.w3.org/ns/did/v1"},
144+			"id":       ident.ServiceDID(s.ForgeHost),
145+			"service": []map[string]string{{
146+				"id": "#tangled_knot", "type": "TangledKnot", "serviceEndpoint": "https://" + s.ForgeHost,
147+			}},
148+		})
149+	})
150+	mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, r *http.Request) {
151+		if _, err := s.Log.List(0, 1); err != nil {
152+			http.Error(w, err.Error(), http.StatusServiceUnavailable)
153+			return
154+		}
155+		fmt.Fprintln(w, "ok")
156+	})
157+	return mux
158+}
159+
160+// Notify is the hook side: spool a push, then poke the server (best
161+// effort). Errors are returned for logging; the hook still exits 0.
162+func Notify(ctx context.Context, spoolDir, server, token string, p Push) error {
163+	if len(p.Updates) == 0 {
164+		return nil
165+	}
166+	if _, err := Spool(spoolDir, p); err != nil {
167+		return fmt.Errorf("spool: %w", err)
168+	}
169+	if server == "" {
170+		return nil
171+	}
172+	ctx, cancel := context.WithTimeout(ctx, 2*time.Second)
173+	defer cancel()
174+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimSuffix(server, "/")+"/ingest", nil)
175+	if err != nil {
176+		return err
177+	}
178+	if token != "" {
179+		req.Header.Set("Authorization", "Bearer "+token)
180+	}
181+	resp, err := http.DefaultClient.Do(req)
182+	if err != nil {
183+		return nil // spooled; the server picks it up on its next pass
184+	}
185+	resp.Body.Close()
186+	return nil
187+}

  1@@ -0,0 +1,287 @@
  2+// Package eventlog is a Tangled-compatible event stream: an SQLite log of
  3+// {rkey, nsid, event, created} served on a websocket as upstream's
  4+// eventstream does (knot and spindle /events). created is a monotonic local
  5+// clock (UnixNano) and is the cursor: GET /events?cursor=N replays events
  6+// with created > N as JSON text messages, then tails.
  7+//
  8+// The bridge serves sh.tangled.git.refUpdate on it; the runner serves
  9+// sh.tangled.pipeline, sh.tangled.pipeline.status and run.rtw.ci.* records.
 10+package eventlog
 11+
 12+import (
 13+	"context"
 14+	"crypto/subtle"
 15+	"database/sql"
 16+	"encoding/json"
 17+	"fmt"
 18+	"net/http"
 19+	"strconv"
 20+	"sync"
 21+	"time"
 22+
 23+	"github.com/gorilla/websocket"
 24+	_ "modernc.org/sqlite"
 25+
 26+	"git.kilimanjaro.io/ci/internal/lex"
 27+	"git.kilimanjaro.io/ci/internal/tid"
 28+)
 29+
 30+const (
 31+	batchSize         = 500
 32+	keepAliveInterval = 30 * time.Second
 33+	writeDeadline     = 10 * time.Second
 34+)
 35+
 36+// Log is an append-only event log.
 37+type Log struct {
 38+	db *sql.DB
 39+
 40+	mu   sync.Mutex
 41+	last int64
 42+	subs map[chan struct{}]struct{}
 43+}
 44+
 45+// Open opens (or creates) the log at path.
 46+func Open(path string) (*Log, error) {
 47+	db, err := sql.Open("sqlite", "file:"+path+"?_pragma=journal_mode(WAL)&_pragma=busy_timeout(5000)&_pragma=synchronous(NORMAL)")
 48+	if err != nil {
 49+		return nil, err
 50+	}
 51+	db.SetMaxOpenConns(1) // one writer; reads are fast and serialised with it
 52+	if _, err := db.Exec(`create table if not exists events (
 53+		created integer primary key,
 54+		rkey text not null,
 55+		nsid text not null,
 56+		event text not null
 57+	)`); err != nil {
 58+		db.Close()
 59+		return nil, err
 60+	}
 61+	l := &Log{db: db, subs: map[chan struct{}]struct{}{}}
 62+	if err := db.QueryRow(`select coalesce(max(created), 0) from events`).Scan(&l.last); err != nil {
 63+		db.Close()
 64+		return nil, err
 65+	}
 66+	return l, nil
 67+}
 68+
 69+// DB exposes the database for packages that keep their own tables next to
 70+// the log (the runner's cursor and pipelines).
 71+func (l *Log) DB() *sql.DB { return l.db }
 72+
 73+// Close closes the database.
 74+func (l *Log) Close() error { return l.db.Close() }
 75+
 76+// Append adds a record (marshalled to JSON) with a new TID rkey unless rkey
 77+// is given, and wakes subscribers.
 78+func (l *Log) Append(nsid, rkey string, record any) (lex.Event, error) {
 79+	b, err := json.Marshal(record)
 80+	if err != nil {
 81+		return lex.Event{}, err
 82+	}
 83+	if rkey == "" {
 84+		rkey = tid.New()
 85+	}
 86+	l.mu.Lock()
 87+	created := time.Now().UnixNano()
 88+	if created <= l.last {
 89+		created = l.last + 1
 90+	}
 91+	_, err = l.db.Exec(`insert into events (created, rkey, nsid, event) values (?, ?, ?, ?)`, created, rkey, nsid, string(b))
 92+	if err == nil {
 93+		l.last = created
 94+	}
 95+	subs := make([]chan struct{}, 0, len(l.subs))
 96+	for c := range l.subs {
 97+		subs = append(subs, c)
 98+	}
 99+	l.mu.Unlock()
100+	if err != nil {
101+		return lex.Event{}, err
102+	}
103+	for _, c := range subs {
104+		select {
105+		case c <- struct{}{}:
106+		default:
107+		}
108+	}
109+	return lex.Event{Rkey: rkey, Nsid: nsid, Event: b, Created: created}, nil
110+}
111+
112+// List returns up to limit events after cursor, oldest first.
113+func (l *Log) List(cursor int64, limit int) ([]lex.Event, error) {
114+	rows, err := l.db.Query(`select created, rkey, nsid, event from events where created > ? order by created limit ?`, cursor, limit)
115+	if err != nil {
116+		return nil, err
117+	}
118+	defer rows.Close()
119+	var out []lex.Event
120+	for rows.Next() {
121+		var ev lex.Event
122+		var body string
123+		if err := rows.Scan(&ev.Created, &ev.Rkey, &ev.Nsid, &body); err != nil {
124+			return nil, err
125+		}
126+		ev.Event = json.RawMessage(body)
127+		out = append(out, ev)
128+	}
129+	return out, rows.Err()
130+}
131+
132+// subscribe returns a channel poked after every append.
133+func (l *Log) subscribe() (chan struct{}, func()) {
134+	c := make(chan struct{}, 1)
135+	l.mu.Lock()
136+	l.subs[c] = struct{}{}
137+	l.mu.Unlock()
138+	return c, func() {
139+		l.mu.Lock()
140+		delete(l.subs, c)
141+		l.mu.Unlock()
142+	}
143+}
144+
145+var upgrader = websocket.Upgrader{ReadBufferSize: 1024, WriteBufferSize: 1024}
146+
147+// ServeHTTP serves the stream (GET /events?cursor=N).
148+func (l *Log) ServeHTTP(w http.ResponseWriter, r *http.Request) {
149+	var cursor int64
150+	if raw := r.URL.Query().Get("cursor"); raw != "" {
151+		c, err := strconv.ParseInt(raw, 10, 64)
152+		if err != nil {
153+			http.Error(w, "invalid cursor", http.StatusBadRequest)
154+			return
155+		}
156+		cursor = c
157+	}
158+	conn, err := upgrader.Upgrade(w, r, nil)
159+	if err != nil {
160+		return
161+	}
162+	defer conn.Close()
163+	poke, unsubscribe := l.subscribe()
164+	defer unsubscribe()
165+
166+	ctx, cancel := context.WithCancel(r.Context())
167+	defer cancel()
168+	go func() {
169+		for {
170+			if _, _, err := conn.NextReader(); err != nil {
171+				cancel()
172+				return
173+			}
174+		}
175+	}()
176+	drain := func() error {
177+		for {
178+			evs, err := l.List(cursor, batchSize)
179+			if err != nil {
180+				return err
181+			}
182+			for _, ev := range evs {
183+				b, _ := json.Marshal(ev)
184+				_ = conn.SetWriteDeadline(time.Now().Add(writeDeadline))
185+				if err := conn.WriteMessage(websocket.TextMessage, b); err != nil {
186+					return err
187+				}
188+				cursor = ev.Created
189+			}
190+			if len(evs) < batchSize {
191+				return nil
192+			}
193+		}
194+	}
195+	if drain() != nil {
196+		return
197+	}
198+	for {
199+		select {
200+		case <-ctx.Done():
201+			return
202+		case <-poke:
203+			if drain() != nil {
204+				return
205+			}
206+		case <-time.After(keepAliveInterval):
207+			if conn.WriteControl(websocket.PingMessage, nil, time.Now().Add(writeDeadline)) != nil {
208+				return
209+			}
210+		}
211+	}
212+}
213+
214+// Subscribe follows a remote stream from cursor, calling f for each event in
215+// order, reconnecting with backoff until ctx ends. f's error stops the
216+// subscription (return it to retry later from the last good cursor). token,
217+// if set, is sent as a bearer token.
218+func Subscribe(ctx context.Context, url string, cursor int64, token string, f func(lex.Event) error) error {
219+	backoff := time.Second
220+	for {
221+		err := follow(ctx, url, &cursor, token, f)
222+		if ctx.Err() != nil {
223+			return ctx.Err()
224+		}
225+		if fe, ok := err.(handlerError); ok {
226+			return fe.err
227+		}
228+		select {
229+		case <-ctx.Done():
230+			return ctx.Err()
231+		case <-time.After(backoff):
232+		}
233+		backoff = min(backoff*2, time.Minute)
234+	}
235+}
236+
237+type handlerError struct{ err error }
238+
239+func (h handlerError) Error() string { return h.err.Error() }
240+
241+func follow(ctx context.Context, url string, cursor *int64, token string, f func(lex.Event) error) error {
242+	h := http.Header{}
243+	if token != "" {
244+		h.Set("Authorization", "Bearer "+token)
245+	}
246+	conn, resp, err := websocket.DefaultDialer.DialContext(ctx, fmt.Sprintf("%s?cursor=%d", url, *cursor), h)
247+	if err != nil {
248+		if resp != nil {
249+			return fmt.Errorf("%s: %s", url, resp.Status)
250+		}
251+		return err
252+	}
253+	defer conn.Close()
254+	go func() {
255+		<-ctx.Done()
256+		conn.Close()
257+	}()
258+	for {
259+		_, msg, err := conn.ReadMessage()
260+		if err != nil {
261+			return err
262+		}
263+		var ev lex.Event
264+		if err := json.Unmarshal(msg, &ev); err != nil {
265+			return err
266+		}
267+		if err := f(ev); err != nil {
268+			return handlerError{err}
269+		}
270+		*cursor = ev.Created
271+	}
272+}
273+
274+// RequireToken wraps h so that requests must carry "Authorization: Bearer
275+// <token>". An empty token disables the check.
276+func RequireToken(token string, h http.Handler) http.Handler {
277+	if token == "" {
278+		return h
279+	}
280+	want := []byte("Bearer " + token)
281+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
282+		if subtle.ConstantTimeCompare([]byte(r.Header.Get("Authorization")), want) != 1 {
283+			http.Error(w, "unauthorized", http.StatusUnauthorized)
284+			return
285+		}
286+		h.ServeHTTP(w, r)
287+	})
288+}

  1@@ -0,0 +1,125 @@
  2+package eventlog
  3+
  4+import (
  5+	"context"
  6+	"net/http/httptest"
  7+	"path/filepath"
  8+	"strings"
  9+	"sync"
 10+	"testing"
 11+	"time"
 12+
 13+	"git.kilimanjaro.io/ci/internal/lex"
 14+)
 15+
 16+func open(t *testing.T) *Log {
 17+	t.Helper()
 18+	l, err := Open(filepath.Join(t.TempDir(), "events.db"))
 19+	if err != nil {
 20+		t.Fatal(err)
 21+	}
 22+	t.Cleanup(func() { l.Close() })
 23+	return l
 24+}
 25+
 26+func TestAppendListMonotonic(t *testing.T) {
 27+	l := open(t)
 28+	var prev int64
 29+	for i := range 100 {
 30+		ev, err := l.Append("x.test", "", map[string]int{"i": i})
 31+		if err != nil {
 32+			t.Fatal(err)
 33+		}
 34+		if ev.Created <= prev {
 35+			t.Fatalf("created not increasing: %d after %d", ev.Created, prev)
 36+		}
 37+		prev = ev.Created
 38+	}
 39+	all, _ := l.List(0, 1000)
 40+	if len(all) != 100 {
 41+		t.Fatalf("%d events", len(all))
 42+	}
 43+	tail, _ := l.List(all[89].Created, 1000)
 44+	if len(tail) != 10 || string(tail[0].Event) != `{"i":90}` {
 45+		t.Errorf("after cursor: %d %s", len(tail), tail[0].Event)
 46+	}
 47+	// reopening keeps the clock ahead of what's stored
 48+	path := filepath.Join(t.TempDir(), "e.db")
 49+	a, _ := Open(path)
 50+	e1, _ := a.Append("x", "", 1)
 51+	a.Close()
 52+	b, _ := Open(path)
 53+	defer b.Close()
 54+	if b.last != e1.Created {
 55+		t.Errorf("reopened last %d want %d", b.last, e1.Created)
 56+	}
 57+}
 58+
 59+func TestStreamReplayTailAndResume(t *testing.T) {
 60+	l := open(t)
 61+	for i := range 3 {
 62+		l.Append("x.test", "", i)
 63+	}
 64+	srv := httptest.NewServer(RequireToken("s3cret", l))
 65+	defer srv.Close()
 66+	url := "ws" + strings.TrimPrefix(srv.URL, "http")
 67+
 68+	var mu sync.Mutex
 69+	var got []lex.Event
 70+	ctx, cancel := context.WithCancel(context.Background())
 71+	done := make(chan error)
 72+	go func() {
 73+		done <- Subscribe(ctx, url, 0, "s3cret", func(ev lex.Event) error {
 74+			mu.Lock()
 75+			got = append(got, ev)
 76+			mu.Unlock()
 77+			return nil
 78+		})
 79+	}()
 80+	wait := func(n int) {
 81+		t.Helper()
 82+		for range 200 {
 83+			mu.Lock()
 84+			k := len(got)
 85+			mu.Unlock()
 86+			if k >= n {
 87+				return
 88+			}
 89+			time.Sleep(10 * time.Millisecond)
 90+		}
 91+		t.Fatalf("got %d events, want %d", len(got), n)
 92+	}
 93+	wait(3)
 94+	l.Append("x.test", "", 3) // tail
 95+	wait(4)
 96+	cancel()
 97+	<-done
 98+
 99+	// resume from the last cursor: only newer events
100+	l.Append("x.test", "", 4)
101+	var resumed []lex.Event
102+	ctx2, cancel2 := context.WithCancel(context.Background())
103+	go func() {
104+		_ = Subscribe(ctx2, url, got[3].Created, "s3cret", func(ev lex.Event) error {
105+			resumed = append(resumed, ev)
106+			cancel2()
107+			return nil
108+		})
109+	}()
110+	<-ctx2.Done()
111+	time.Sleep(20 * time.Millisecond)
112+	if len(resumed) != 1 || string(resumed[0].Event) != "4" {
113+		t.Errorf("resumed %+v", resumed)
114+	}
115+}
116+
117+func TestStreamNeedsToken(t *testing.T) {
118+	l := open(t)
119+	srv := httptest.NewServer(RequireToken("s3cret", l))
120+	defer srv.Close()
121+	url := "ws" + strings.TrimPrefix(srv.URL, "http")
122+	var c int64
123+	if err := follow(context.Background(), url, &c, "wrong", func(lex.Event) error { return nil }); err == nil || !strings.Contains(err.Error(), "401") {
124+		t.Errorf("wrong token: %v", err)
125+	}
126+}

 1@@ -0,0 +1,34 @@
 2+// Package ident maps forge repositories to the DIDs that identify them in
 3+// Tangled records: did:web:<forge-host>:<repo> (DESIGN.md §4).
 4+package ident
 5+
 6+import (
 7+	"fmt"
 8+	"regexp"
 9+	"strings"
10+)
11+
12+var repoName = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._-]*$`)
13+
14+// RepoDID returns the DID of repo on the forge at host.
15+func RepoDID(host, repo string) string {
16+	return "did:web:" + strings.ReplaceAll(host, ":", "%3A") + ":" + repo
17+}
18+
19+// ParseRepoDID splits a repo DID into forge host and repo name.
20+func ParseRepoDID(did string) (host, repo string, err error) {
21+	rest, ok := strings.CutPrefix(did, "did:web:")
22+	if !ok {
23+		return "", "", fmt.Errorf("%q is not a did:web", did)
24+	}
25+	host, repo, ok = strings.Cut(rest, ":")
26+	if !ok || host == "" || !repoName.MatchString(repo) {
27+		return "", "", fmt.Errorf("%q is not a repo DID (did:web:<host>:<repo>)", did)
28+	}
29+	return strings.ReplaceAll(host, "%3A", ":"), repo, nil
30+}
31+
32+// ServiceDID returns the did:web of a service at host.
33+func ServiceDID(host string) string {
34+	return "did:web:" + strings.ReplaceAll(host, ":", "%3A")
35+}

 1@@ -0,0 +1,22 @@
 2+package ident
 3+
 4+import "testing"
 5+
 6+func TestRepoDID(t *testing.T) {
 7+	d := RepoDID("git.kilimanjaro.io", "sovrn")
 8+	if d != "did:web:git.kilimanjaro.io:sovrn" {
 9+		t.Fatal(d)
10+	}
11+	h, r, err := ParseRepoDID(d)
12+	if err != nil || h != "git.kilimanjaro.io" || r != "sovrn" {
13+		t.Fatal(h, r, err)
14+	}
15+	if h, r, _ := ParseRepoDID(RepoDID("localhost:8080", "x")); h != "localhost:8080" || r != "x" {
16+		t.Error(h, r)
17+	}
18+	for _, bad := range []string{"did:plc:abc", "did:web:host", "did:web:host:a:b", "did:web::x"} {
19+		if _, _, err := ParseRepoDID(bad); err == nil {
20+			t.Errorf("%s accepted", bad)
21+		}
22+	}
23+}