feat: JetStream event caching and management

docs/roadmap.md

···
 - [x] Reuse the existing normalization and upsert path for on-demand indexing jobs
 - [x] Trigger indexing jobs from repo, issue, PR, profile, and similar fetch handlers
 - [x] Add dedupe, retries, and observability for indexing jobs
-- [ ] Add a JetStream cache consumer with a persisted timestamp cursor
-- [ ] Seed the JetStream cursor to `now - 24h` on first boot and rewind slightly on reconnect
-- [ ] Store and serve bounded recent activity from the local cache
+- [x] Add a JetStream cache consumer with a persisted timestamp cursor
+- [x] Seed the JetStream cursor to `now - 24h` on first boot and rewind slightly on reconnect
+- [x] Store and serve bounded recent activity from the local cache
 - [ ] Keep Tap as the authoritative indexing and bulk backfill path
 - [ ] Define a controlled backfill and repo-resync playbook for recovery (`docs/references/resync.md`)
 

packages/api/internal/api/actors.go

···
 	}
 }
 
-// isError is a type-safe errors.As replacement for pointer receiver targets.
-func isError[T error](err error, target *T) bool {
-	if err == nil {
-		return false
-	}
-
-	type unwrapper interface{ Unwrap() error }
-	for e := err; e != nil; {
-		if t, ok := e.(T); ok {
-			*target = t
-			return true
-		}
-		if u, ok := e.(unwrapper); ok {
-			e = u.Unwrap()
-		} else {
-			break
-		}
-	}
-	return false
-}
-
 // handleGetActor returns the actor's Tangled profile + optional Bluesky info.
 // GET /actors/{handle}
 func (s *Server) handleGetActor(w http.ResponseWriter, r *http.Request) {
···
 	return pulls, statusMap, nil
 }
 
-func resolveIssueState(stateMap map[string]string, issueURI string) string {
-	raw := stateMap[issueURI]
-	if strings.HasSuffix(raw, ".closed") {
-		return "closed"
-	}
-	return "open"
-}
-
-func resolvePullStatus(statusMap map[string]string, pullURI string) string {
-	raw := statusMap[pullURI]
-	switch {
-	case strings.HasSuffix(raw, ".merged"):
-		return "merged"
-	case strings.HasSuffix(raw, ".closed"):
-		return "closed"
-	default:
-		return "open"
-	}
-}
-
 type bskyProfileResponse struct {
 	DisplayName string `json:"displayName,omitempty"`
 	Avatar      string `json:"avatar,omitempty"`
···
 	}
 	return &p
 }
-
-// parseATURI splits an AT URI (at://did/collection/rkey) into its components.
-func parseATURI(uri string) (did, collection, rkey string, err error) {
-	trimmed := strings.TrimPrefix(uri, "at://")
-	parts := strings.SplitN(trimmed, "/", 3)
-	if len(parts) != 3 {
-		return "", "", "", fmt.Errorf("invalid AT URI: %q", uri)
-	}
-	return parts[0], parts[1], parts[2], nil
-}

packages/api/internal/api/api.go

···
 	"log/slog"
 	"net"
 	"net/http"
-	"strconv"
 	"strings"
 	"sync"
 	"time"
···
 }
 
 // Handler returns the HTTP handler with all routes registered.
+//
+// TODO: refactor this to have a func router() that returns [http.Handler] (mux),
+// then registerXXX routes
 func (s *Server) Handler() http.Handler {
 	mux := http.NewServeMux()
 
···
 	mux.HandleFunc("GET /profiles/{did}/summary", s.handleProfileSummary)
 
 	mux.HandleFunc("GET /backlinks/count", s.handleBacklinksCount)
+	mux.HandleFunc("GET /activity", s.handleActivity)
 	mux.HandleFunc("GET /activity/stream", s.handleActivityStream)
 	mux.HandleFunc("GET /identity/resolve", s.handleResolveHandle)
 	mux.HandleFunc("GET /identity/did/{did}", s.handleDidDocument)
···
 
 	errCh := make(chan error, 1)
 	go s.runReadThroughIndexer(ctx)
+	go s.runJetstreamConsumer(ctx)
 
 	go func() {
 		s.log.Info("listening", slog.String("addr", s.cfg.HTTPBindAddr))
···
 	writeJSON(w, http.StatusOK, summary)
 }
 
-func (s *Server) handleNotImplemented(w http.ResponseWriter, _ *http.Request) {
-	writeJSON(w, http.StatusNotImplemented, errorBody("not_implemented", "this endpoint is not yet available"))
+// knownActivityParams is the whitelist of accepted query parameters for the activity endpoint.
+var knownActivityParams = map[string]bool{
+	"limit": true, "offset": true,
+	"collection": true, "operation": true, "did": true,
 }
 
-func writeJSON(w http.ResponseWriter, status int, v any) {
-	w.Header().Set("Content-Type", "application/json")
-	w.WriteHeader(status)
-	_ = json.NewEncoder(w).Encode(v)
-}
+// handleActivity returns bounded recent activity from the JetStream cache.
+// Route: GET /activity
+func (s *Server) handleActivity(w http.ResponseWriter, r *http.Request) {
+	for key := range r.URL.Query() {
+		if !knownActivityParams[key] {
+			writeJSON(w, http.StatusBadRequest, errorBody("unknown_parameter", fmt.Sprintf("unknown parameter: %s", key)))
+			return
+		}
+	}
 
-func errorBody(code, message string) map[string]string {
-	return map[string]string{"error": code, "message": message}
-}
+	limit, err := intParam(r, "limit", 50)
+	if err != nil || limit < 1 || limit > 200 {
+		writeJSON(w, http.StatusBadRequest, errorBody("invalid_parameter", "limit must be between 1 and 200"))
+		return
+	}
+	offset, err := intParam(r, "offset", 0)
+	if err != nil || offset < 0 {
+		writeJSON(w, http.StatusBadRequest, errorBody("invalid_parameter", "offset must be >= 0"))
+		return
+	}
 
-func intParam(r *http.Request, key string, def int) (int, error) {
-	v := r.URL.Query().Get(key)
-	if v == "" {
-		return def, nil
+	filter := store.JetstreamEventFilter{
+		Collection: r.URL.Query().Get("collection"),
+		Operation:  r.URL.Query().Get("operation"),
+		DID:        r.URL.Query().Get("did"),
+		Limit:      limit,
+		Offset:     offset,
 	}
-	n, err := strconv.Atoi(v)
+
+	events, err := s.store.ListJetstreamEvents(r.Context(), filter)
 	if err != nil {
-		return 0, err
+		s.log.Error("list activity failed", slog.String("error", err.Error()))
+		writeJSON(w, http.StatusInternalServerError, errorBody("db_error", "failed to fetch activity"))
+		return
 	}
-	return n, nil
-}
 
-type documentJSON struct {
-	ID           string `json:"id"`
-	DID          string `json:"did"`
-	Collection   string `json:"collection"`
-	RKey         string `json:"rkey"`
-	ATURI        string `json:"at_uri"`
-	CID          string `json:"cid"`
-	RecordType   string `json:"record_type"`
-	Title        string `json:"title"`
-	Body         string `json:"body"`
-	Summary      string `json:"summary,omitempty"`
-	RepoName     string `json:"repo_name,omitempty"`
-	AuthorHandle string `json:"author_handle,omitempty"`
-	TagsJSON     string `json:"tags_json,omitempty"`
-	Language     string `json:"language,omitempty"`
-	WebURL       string `json:"web_url,omitempty"`
-	CreatedAt    string `json:"created_at,omitempty"`
-	UpdatedAt    string `json:"updated_at,omitempty"`
-	IndexedAt    string `json:"indexed_at"`
+	type eventJSON struct {
+		TimeUS     int64           `json:"time_us"`
+		DID        string          `json:"did"`
+		Kind       string          `json:"kind"`
+		Collection string          `json:"collection,omitempty"`
+		RKey       string          `json:"rkey,omitempty"`
+		Operation  string          `json:"operation,omitempty"`
+		Payload    json.RawMessage `json:"payload"`
+	}
+
+	out := make([]eventJSON, 0, len(events))
+	for _, e := range events {
+		out = append(out, eventJSON{
+			TimeUS:     e.TimeUS,
+			DID:        e.DID,
+			Kind:       e.Kind,
+			Collection: e.Collection,
+			RKey:       e.RKey,
+			Operation:  e.Operation,
+			Payload:    json.RawMessage(e.Payload),
+		})
+	}
+
+	writeJSON(w, http.StatusOK, map[string]any{
+		"limit":  limit,
+		"offset": offset,
+		"events": out,
+	})
 }
 
-func documentResponse(doc *store.Document) documentJSON {
-	return documentJSON{
-		ID:           doc.ID,
-		DID:          doc.DID,
-		Collection:   doc.Collection,
-		RKey:         doc.RKey,
-		ATURI:        doc.ATURI,
-		CID:          doc.CID,
-		RecordType:   doc.RecordType,
-		Title:        doc.Title,
-		Body:         doc.Body,
-		Summary:      doc.Summary,
-		RepoName:     doc.RepoName,
-		AuthorHandle: doc.AuthorHandle,
-		TagsJSON:     doc.TagsJSON,
-		Language:     doc.Language,
-		WebURL:       doc.WebURL,
-		CreatedAt:    doc.CreatedAt,
-		UpdatedAt:    doc.UpdatedAt,
-		IndexedAt:    doc.IndexedAt,
-	}
+func (s *Server) handleNotImplemented(w http.ResponseWriter, _ *http.Request) {
+	writeJSON(w, http.StatusNotImplemented, errorBody("not_implemented", "this endpoint is not yet available"))
 }

packages/api/internal/api/helpers.go

···
+package api
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"strconv"
+	"strings"
+	"time"
+
+	"tangled.org/desertthunder.dev/twister/internal/store"
+)
+
+func writeJSON(w http.ResponseWriter, status int, v any) {
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(status)
+	_ = json.NewEncoder(w).Encode(v)
+}
+
+func errorBody(code, message string) map[string]string {
+	return map[string]string{"error": code, "message": message}
+}
+
+func intParam(r *http.Request, key string, def int) (int, error) {
+	v := r.URL.Query().Get(key)
+	if v == "" {
+		return def, nil
+	}
+	n, err := strconv.Atoi(v)
+	if err != nil {
+		return 0, err
+	}
+	return n, nil
+}
+
+type documentJSON struct {
+	ID           string `json:"id"`
+	DID          string `json:"did"`
+	Collection   string `json:"collection"`
+	RKey         string `json:"rkey"`
+	ATURI        string `json:"at_uri"`
+	CID          string `json:"cid"`
+	RecordType   string `json:"record_type"`
+	Title        string `json:"title"`
+	Body         string `json:"body"`
+	Summary      string `json:"summary,omitempty"`
+	RepoName     string `json:"repo_name,omitempty"`
+	AuthorHandle string `json:"author_handle,omitempty"`
+	TagsJSON     string `json:"tags_json,omitempty"`
+	Language     string `json:"language,omitempty"`
+	WebURL       string `json:"web_url,omitempty"`
+	CreatedAt    string `json:"created_at,omitempty"`
+	UpdatedAt    string `json:"updated_at,omitempty"`
+	IndexedAt    string `json:"indexed_at"`
+}
+
+func documentResponse(doc *store.Document) documentJSON {
+	return documentJSON{
+		ID:           doc.ID,
+		DID:          doc.DID,
+		Collection:   doc.Collection,
+		RKey:         doc.RKey,
+		ATURI:        doc.ATURI,
+		CID:          doc.CID,
+		RecordType:   doc.RecordType,
+		Title:        doc.Title,
+		Body:         doc.Body,
+		Summary:      doc.Summary,
+		RepoName:     doc.RepoName,
+		AuthorHandle: doc.AuthorHandle,
+		TagsJSON:     doc.TagsJSON,
+		Language:     doc.Language,
+		WebURL:       doc.WebURL,
+		CreatedAt:    doc.CreatedAt,
+		UpdatedAt:    doc.UpdatedAt,
+		IndexedAt:    doc.IndexedAt,
+	}
+}
+
+// isError is a type-safe errors.As replacement for pointer receiver targets.
+func isError[T error](err error, target *T) bool {
+	if err == nil {
+		return false
+	}
+
+	type unwrapper interface{ Unwrap() error }
+	for e := err; e != nil; {
+		if t, ok := e.(T); ok {
+			*target = t
+			return true
+		}
+		if u, ok := e.(unwrapper); ok {
+			e = u.Unwrap()
+		} else {
+			break
+		}
+	}
+	return false
+}
+
+func resolveIssueState(stateMap map[string]string, issueURI string) string {
+	raw := stateMap[issueURI]
+	if strings.HasSuffix(raw, ".closed") {
+		return "closed"
+	}
+	return "open"
+}
+
+func resolvePullStatus(statusMap map[string]string, pullURI string) string {
+	raw := statusMap[pullURI]
+	switch {
+	case strings.HasSuffix(raw, ".merged"):
+		return "merged"
+	case strings.HasSuffix(raw, ".closed"):
+		return "closed"
+	default:
+		return "open"
+	}
+}
+
+// parseATURI splits an AT URI (at://did/collection/rkey) into its components.
+func parseATURI(uri string) (did, collection, rkey string, err error) {
+	trimmed := strings.TrimPrefix(uri, "at://")
+	parts := strings.SplitN(trimmed, "/", 3)
+	if len(parts) != 3 {
+		return "", "", "", fmt.Errorf("invalid AT URI: %q", uri)
+	}
+	return parts[0], parts[1], parts[2], nil
+}
+
+func retryDelay(attempt int) time.Duration {
+	if attempt < 1 {
+		attempt = 1
+	}
+	base := time.Second * time.Duration(1<<min(attempt-1, 8))
+	if base > 5*time.Minute {
+		return 5 * time.Minute
+	}
+	return base
+}
+
+func truncateErr(err error) string {
+	if err == nil {
+		return ""
+	}
+	msg := err.Error()
+	if len(msg) > 500 {
+		return msg[:500]
+	}
+	return msg
+}

packages/api/internal/api/jetstream.go

···
+package api
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log/slog"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/coder/websocket"
+	"tangled.org/desertthunder.dev/twister/internal/store"
+)
+
+const (
+	jetstreamConsumerName   = "jetstream-cache-v1"
+	jetstreamReconnectDelay = 5 * time.Second
+	// persist cursor every N events
+	jetstreamCursorInterval = 50
+)
+
+type jetstreamMessage struct {
+	DID    string `json:"did"`
+	TimeUS int64  `json:"time_us"`
+	Kind   string `json:"kind"`
+	Commit *struct {
+		Operation  string `json:"operation"`
+		Collection string `json:"collection"`
+		RKey       string `json:"rkey"`
+	} `json:"commit"`
+}
+
+// runJetstreamConsumer connects to JetStream and caches sh.tangled.* events.
+// Reconnects automatically after any disconnect.
+func (s *Server) runJetstreamConsumer(ctx context.Context) {
+	s.log.Info("jetstream cache consumer starting")
+	for {
+		if ctx.Err() != nil {
+			s.log.Info("jetstream cache consumer stopped")
+			return
+		}
+		if err := s.consumeJetstream(ctx); err != nil && ctx.Err() == nil {
+			s.log.Warn("jetstream consumer disconnected; reconnecting",
+				slog.String("error", err.Error()),
+				slog.Duration("retry_in", jetstreamReconnectDelay),
+			)
+		}
+		select {
+		case <-ctx.Done():
+			return
+		case <-time.After(jetstreamReconnectDelay):
+		}
+	}
+}
+
+func (s *Server) consumeJetstream(ctx context.Context) error {
+	cursorUS, err := s.loadJetstreamCursor(ctx)
+	if err != nil {
+		return fmt.Errorf("load jetstream cursor: %w", err)
+	}
+
+	u := buildJetstreamURL(s.cfg.JetstreamURL, s.cfg.JetstreamWantedCollections, cursorUS)
+	conn, _, err := websocket.Dial(ctx, u, nil)
+	if err != nil {
+		return fmt.Errorf("jetstream dial: %w", err)
+	}
+	defer conn.CloseNow()
+
+	s.log.Info("jetstream consumer connected", slog.Int64("cursor_us", cursorUS))
+
+	var count int64
+	var lastTimeUS int64 = cursorUS
+
+	for {
+		if ctx.Err() != nil {
+			return nil
+		}
+
+		_, msg, err := conn.Read(ctx)
+		if err != nil {
+			if ctx.Err() != nil {
+				return nil
+			}
+			return fmt.Errorf("jetstream read: %w", err)
+		}
+
+		var jmsg jetstreamMessage
+		if err := json.Unmarshal(msg, &jmsg); err != nil {
+			s.log.Debug("jetstream parse failed", slog.String("error", err.Error()))
+			continue
+		}
+		if jmsg.DID == "" || jmsg.TimeUS == 0 {
+			continue
+		}
+
+		evt := &store.JetstreamEvent{
+			TimeUS:     jmsg.TimeUS,
+			DID:        jmsg.DID,
+			Kind:       jmsg.Kind,
+			Payload:    string(msg),
+			ReceivedAt: time.Now().UTC().Format(time.RFC3339),
+		}
+		if jmsg.Commit != nil {
+			evt.Collection = jmsg.Commit.Collection
+			evt.RKey = jmsg.Commit.RKey
+			evt.Operation = jmsg.Commit.Operation
+		}
+
+		if err := s.store.InsertJetstreamEvent(ctx, evt, s.cfg.ActivityMaxEvents); err != nil {
+			s.log.Warn("jetstream insert failed", slog.String("error", err.Error()))
+		}
+
+		lastTimeUS = jmsg.TimeUS
+		count++
+
+		if count%jetstreamCursorInterval == 0 {
+			cursor := strconv.FormatInt(lastTimeUS, 10)
+			if err := s.store.SetSyncState(ctx, jetstreamConsumerName, cursor); err != nil {
+				s.log.Warn("jetstream cursor persist failed", slog.String("error", err.Error()))
+			}
+		}
+	}
+}
+
+// loadJetstreamCursor returns the cursor to use when connecting to JetStream.
+// First boot: seeds to now-24h. Subsequent connects: rewinds by ActivityRewindDuration.
+func (s *Server) loadJetstreamCursor(ctx context.Context) (int64, error) {
+	state, err := s.store.GetSyncState(ctx, jetstreamConsumerName)
+	if err != nil {
+		return 0, err
+	}
+
+	if state == nil || strings.TrimSpace(state.Cursor) == "" {
+		seed := time.Now().UTC().Add(-24 * time.Hour).UnixMicro()
+		s.log.Info("jetstream cursor: first boot, seeding to now-24h", slog.Int64("cursor_us", seed))
+		return seed, nil
+	}
+
+	cursor, err := strconv.ParseInt(strings.TrimSpace(state.Cursor), 10, 64)
+	if err != nil {
+		seed := time.Now().UTC().Add(-24 * time.Hour).UnixMicro()
+		s.log.Warn("jetstream cursor parse failed; reseeding to now-24h",
+			slog.String("raw", state.Cursor), slog.String("error", err.Error()))
+		return seed, nil
+	}
+
+	rewound := cursor - s.cfg.ActivityRewindDuration.Microseconds()
+	s.log.Info("jetstream cursor: rewinding on reconnect",
+		slog.Int64("prev_us", cursor),
+		slog.Duration("rewind", s.cfg.ActivityRewindDuration),
+		slog.Int64("cursor_us", rewound),
+	)
+	return rewound, nil
+}
+
+// buildJetstreamURL constructs the JetStream subscribe URL with wanted collections
+// and cursor. The wildcard character in collection names is not percent-encoded.
+func buildJetstreamURL(base, collections string, cursorUS int64) string {
+	var parts []string
+	for _, c := range strings.Split(collections, ",") {
+		c = strings.TrimSpace(c)
+		if c != "" {
+			parts = append(parts, "wantedCollections="+c)
+		}
+	}
+	if cursorUS > 0 {
+		parts = append(parts, "cursor="+strconv.FormatInt(cursorUS, 10))
+	}
+	if len(parts) == 0 {
+		return base
+	}
+	return base + "?" + strings.Join(parts, "&")
+}

packages/api/internal/api/jetstream_test.go

···
+package api
+
+import (
+	"strings"
+	"testing"
+)
+
+func TestBuildJetstreamURL(t *testing.T) {
+	base := "wss://jetstream2.us-east.bsky.network/subscribe"
+
+	t.Run("single collection with cursor", func(t *testing.T) {
+		u := buildJetstreamURL(base, "sh.tangled.*", 1_000_000)
+		if !strings.Contains(u, "wantedCollections=sh.tangled.*") {
+			t.Errorf("missing wantedCollections in %q", u)
+		}
+		if !strings.Contains(u, "cursor=1000000") {
+			t.Errorf("missing cursor in %q", u)
+		}
+	})
+
+	t.Run("multiple collections", func(t *testing.T) {
+		u := buildJetstreamURL(base, "sh.tangled.repo, sh.tangled.repo.issue", 0)
+		if !strings.Contains(u, "wantedCollections=sh.tangled.repo") {
+			t.Errorf("missing first collection in %q", u)
+		}
+		if !strings.Contains(u, "wantedCollections=sh.tangled.repo.issue") {
+			t.Errorf("missing second collection in %q", u)
+		}
+		if strings.Contains(u, "cursor=") {
+			t.Errorf("unexpected cursor when cursorUS=0 in %q", u)
+		}
+	})
+
+	t.Run("no collections no cursor", func(t *testing.T) {
+		u := buildJetstreamURL(base, "", 0)
+		if u != base {
+			t.Errorf("expected bare base URL, got %q", u)
+		}
+	})
+
+	t.Run("wildcard is not percent-encoded", func(t *testing.T) {
+		u := buildJetstreamURL(base, "sh.tangled.*", 42)
+		if strings.Contains(u, "%2A") {
+			t.Errorf("wildcard was percent-encoded in %q", u)
+		}
+	})
+}

packages/api/internal/api/readthrough.go

···
 		s.enqueueXRPCRecord(ctx, e.URI, e.CID, e.Value)
 	}
 }
-
-func retryDelay(attempt int) time.Duration {
-	if attempt < 1 {
-		attempt = 1
-	}
-	base := time.Second * time.Duration(1<<minInt(attempt-1, 8))
-	if base > 5*time.Minute {
-		return 5 * time.Minute
-	}
-	return base
-}
-
-func truncateErr(err error) string {
-	if err == nil {
-		return ""
-	}
-	msg := err.Error()
-	if len(msg) > 500 {
-		return msg[:500]
-	}
-	return msg
-}
-
-func minInt(a, b int) int {
-	if a < b {
-		return a
-	}
-	return b
-}

packages/api/internal/config/config.go

···
 	ConstellationUserAgent string
 	ConstellationTimeout   time.Duration
 	ConstellationCacheTTL  time.Duration
-	OAuthClientID          string
-	OAuthRedirectURIs      []string
+	OAuthClientID             string
+	OAuthRedirectURIs         []string
+	JetstreamURL              string
+	JetstreamWantedCollections string
+	ActivityMaxEvents         int
+	ActivityRewindDuration    time.Duration
 }
 
 type LoadOptions struct {
···
 		ConstellationUserAgent: envOrDefault("CONSTELLATION_USER_AGENT", "twister/1.0 (https://tangled.org/desertthunder.dev/twisted; Owais <[e-mail]>)"),
 		ConstellationTimeout:   envDuration("CONSTELLATION_TIMEOUT", 10*time.Second),
 		ConstellationCacheTTL:  envDuration("CONSTELLATION_CACHE_TTL", 5*time.Minute),
-		OAuthClientID:          os.Getenv("OAUTH_CLIENT_ID"),
-		OAuthRedirectURIs:      envSlice("OAUTH_REDIRECT_URIS", nil),
+		OAuthClientID:              os.Getenv("OAUTH_CLIENT_ID"),
+		OAuthRedirectURIs:          envSlice("OAUTH_REDIRECT_URIS", nil),
+		JetstreamURL:               envOrDefault("JETSTREAM_URL", "wss://jetstream2.us-east.bsky.network/subscribe"),
+		JetstreamWantedCollections: envOrDefault("JETSTREAM_WANTED_COLLECTIONS", "sh.tangled.*"),
+		ActivityMaxEvents:          envInt("ACTIVITY_MAX_EVENTS", 500),
+		ActivityRewindDuration:     envDuration("ACTIVITY_REWIND_DURATION", 5*time.Minute),
 	}
 
 	if opts.Local {

packages/api/internal/ingest/ingest_test.go

···
 	return 0, nil
 }
 
+func (f *fakeStore) InsertJetstreamEvent(_ context.Context, _ *store.JetstreamEvent, _ int) error {
+	return nil
+}
+
+func (f *fakeStore) ListJetstreamEvents(_ context.Context, _ store.JetstreamEventFilter) ([]*store.JetstreamEvent, error) {
+	return nil, nil
+}
+
 func (f *fakeStore) Ping(_ context.Context) error {
 	return nil
 }

packages/api/internal/store/migrations/006_jetstream_events.sql

···
+CREATE TABLE IF NOT EXISTS jetstream_events (
+    id          INTEGER PRIMARY KEY AUTOINCREMENT,
+    time_us     INTEGER NOT NULL,
+    did         TEXT NOT NULL,
+    kind        TEXT NOT NULL,
+    collection  TEXT,
+    rkey        TEXT,
+    operation   TEXT,
+    payload     TEXT NOT NULL,
+    received_at TEXT NOT NULL
+);
+
+CREATE INDEX IF NOT EXISTS idx_jetstream_events_time_us
+    ON jetstream_events(time_us DESC);

packages/api/internal/store/sql_store.go

···
 	return n, nil
 }
 
+func (s *SQLStore) InsertJetstreamEvent(ctx context.Context, event *JetstreamEvent, maxEvents int) error {
+	tx, err := s.db.BeginTx(ctx, nil)
+	if err != nil {
+		return fmt.Errorf("begin insert jetstream event tx: %w", err)
+	}
+	defer tx.Rollback()
+
+	_, err = tx.ExecContext(ctx, `
+		INSERT INTO jetstream_events (time_us, did, kind, collection, rkey, operation, payload, received_at)
+		VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
+		event.TimeUS, event.DID, event.Kind,
+		nullableStr(event.Collection), nullableStr(event.RKey), nullableStr(event.Operation),
+		event.Payload, event.ReceivedAt,
+	)
+	if err != nil {
+		return fmt.Errorf("insert jetstream event: %w", err)
+	}
+
+	if maxEvents > 0 {
+		_, err = tx.ExecContext(ctx, `
+			DELETE FROM jetstream_events
+			WHERE id NOT IN (
+				SELECT id FROM jetstream_events ORDER BY time_us DESC LIMIT ?
+			)`, maxEvents)
+		if err != nil {
+			return fmt.Errorf("trim jetstream events: %w", err)
+		}
+	}
+
+	if err := tx.Commit(); err != nil {
+		return fmt.Errorf("commit insert jetstream event tx: %w", err)
+	}
+	return nil
+}
+
+func (s *SQLStore) ListJetstreamEvents(ctx context.Context, filter JetstreamEventFilter) ([]*JetstreamEvent, error) {
+	query := `
+		SELECT id, time_us, did, kind,
+		       COALESCE(collection, ''), COALESCE(rkey, ''), COALESCE(operation, ''),
+		       payload, received_at
+		FROM jetstream_events WHERE 1=1`
+	args := []any{}
+
+	if filter.Collection != "" {
+		query += " AND collection = ?"
+		args = append(args, filter.Collection)
+	}
+	if filter.DID != "" {
+		query += " AND did = ?"
+		args = append(args, filter.DID)
+	}
+	if filter.Operation != "" {
+		query += " AND operation = ?"
+		args = append(args, filter.Operation)
+	}
+
+	query += " ORDER BY time_us DESC"
+
+	limit := filter.Limit
+	if limit <= 0 {
+		limit = 50
+	}
+	query += " LIMIT ?"
+	args = append(args, limit)
+
+	if filter.Offset > 0 {
+		query += " OFFSET ?"
+		args = append(args, filter.Offset)
+	}
+
+	rows, err := s.db.QueryContext(ctx, query, args...)
+	if err != nil {
+		return nil, fmt.Errorf("list jetstream events: %w", err)
+	}
+	defer rows.Close()
+
+	var events []*JetstreamEvent
+	for rows.Next() {
+		e := &JetstreamEvent{}
+		if err := rows.Scan(
+			&e.ID, &e.TimeUS, &e.DID, &e.Kind,
+			&e.Collection, &e.RKey, &e.Operation,
+			&e.Payload, &e.ReceivedAt,
+		); err != nil {
+			return nil, fmt.Errorf("scan jetstream event: %w", err)
+		}
+		events = append(events, e)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("iterate jetstream events: %w", err)
+	}
+	return events, nil
+}
+
 func (s *SQLStore) Ping(ctx context.Context) error {
 	return s.db.PingContext(ctx)
 }

packages/api/internal/store/store.go

···
 	UpdatedAt   string
 }
 
+// JetstreamEvent is a cached JetStream activity event.
+type JetstreamEvent struct {
+	ID         int64
+	TimeUS     int64
+	DID        string
+	Kind       string
+	Collection string
+	RKey       string
+	Operation  string
+	Payload    string
+	ReceivedAt string
+}
+
+// JetstreamEventFilter scopes a ListJetstreamEvents query.
+type JetstreamEventFilter struct {
+	Collection string
+	DID        string
+	Operation  string
+	Limit      int
+	Offset     int
+}
+
 // IndexingJobInput is the payload used to enqueue or refresh an indexing job.
 type IndexingJobInput struct {
 	DocumentID string
···
 	GetRepoCollaborators(ctx context.Context, repoOwnerDID string) ([]string, error)
 	CountDocuments(ctx context.Context) (int64, error)
 	CountPendingIndexingJobs(ctx context.Context) (int64, error)
+	InsertJetstreamEvent(ctx context.Context, event *JetstreamEvent, maxEvents int) error
+	ListJetstreamEvents(ctx context.Context, filter JetstreamEventFilter) ([]*JetstreamEvent, error)
 	Ping(ctx context.Context) error
 }

packages/api/internal/store/store_test.go

···
 		}
 	})
 
+	t.Run("jetstream events insert and list", func(t *testing.T) {
+		evt := &store.JetstreamEvent{
+			TimeUS:     1_000_000,
+			DID:        "did:plc:evt1",
+			Kind:       "commit",
+			Collection: "sh.tangled.repo",
+			RKey:       "rkey1",
+			Operation:  "create",
+			Payload:    `{"did":"did:plc:evt1","time_us":1000000,"kind":"commit"}`,
+			ReceivedAt: "2026-01-01T00:00:00Z",
+		}
+		if err := st.InsertJetstreamEvent(ctx, evt, 500); err != nil {
+			t.Fatalf("insert jetstream event: %v", err)
+		}
+
+		events, err := st.ListJetstreamEvents(ctx, store.JetstreamEventFilter{Limit: 10})
+		if err != nil {
+			t.Fatalf("list jetstream events: %v", err)
+		}
+		if len(events) < 1 {
+			t.Fatal("expected at least one jetstream event")
+		}
+		found := false
+		for _, e := range events {
+			if e.DID == "did:plc:evt1" && e.Collection == "sh.tangled.repo" {
+				found = true
+				if e.Operation != "create" {
+					t.Errorf("operation: got %q, want create", e.Operation)
+				}
+				if e.TimeUS != 1_000_000 {
+					t.Errorf("time_us: got %d, want 1000000", e.TimeUS)
+				}
+			}
+		}
+		if !found {
+			t.Error("inserted event not found in list")
+		}
+	})
+
+	t.Run("jetstream events filter by collection", func(t *testing.T) {
+		evt := &store.JetstreamEvent{
+			TimeUS:     2_000_000,
+			DID:        "did:plc:evt2",
+			Kind:       "commit",
+			Collection: "sh.tangled.repo.issue",
+			RKey:       "rkey2",
+			Operation:  "create",
+			Payload:    `{"did":"did:plc:evt2","time_us":2000000,"kind":"commit"}`,
+			ReceivedAt: "2026-01-01T00:00:01Z",
+		}
+		if err := st.InsertJetstreamEvent(ctx, evt, 500); err != nil {
+			t.Fatalf("insert jetstream event: %v", err)
+		}
+
+		events, err := st.ListJetstreamEvents(ctx, store.JetstreamEventFilter{
+			Collection: "sh.tangled.repo.issue",
+			Limit:      10,
+		})
+		if err != nil {
+			t.Fatalf("list with collection filter: %v", err)
+		}
+		for _, e := range events {
+			if e.Collection != "sh.tangled.repo.issue" {
+				t.Errorf("filter leaked event with collection %q", e.Collection)
+			}
+		}
+		if len(events) == 0 {
+			t.Error("expected at least one event matching collection filter")
+		}
+	})
+
+	t.Run("jetstream events bounded by maxEvents", func(t *testing.T) {
+		// Insert 5 events with max=3; only the 3 most recent should survive.
+		for i := int64(1); i <= 5; i++ {
+			e := &store.JetstreamEvent{
+				TimeUS:     100 + i,
+				DID:        "did:plc:bound",
+				Kind:       "commit",
+				Collection: "sh.tangled.repo",
+				RKey:       "rkey-bound",
+				Operation:  "create",
+				Payload:    `{}`,
+				ReceivedAt: "2026-01-01T00:00:00Z",
+			}
+			if err := st.InsertJetstreamEvent(ctx, e, 3); err != nil {
+				t.Fatalf("insert bounded event %d: %v", i, err)
+			}
+		}
+
+		all, err := st.ListJetstreamEvents(ctx, store.JetstreamEventFilter{
+			DID:   "did:plc:bound",
+			Limit: 100,
+		})
+		if err != nil {
+			t.Fatalf("list bounded events: %v", err)
+		}
+		// After inserting 5 events total (including the one from the previous subtests),
+		// maxEvents=3 means at most 3 rows survive across all events.
+		if len(all) > 3 {
+			t.Errorf("expected at most 3 events after bounding, got %d", len(all))
+		}
+	})
+
 	t.Run("indexing jobs enqueue claim retry complete", func(t *testing.T) {
 		job := store.IndexingJobInput{
 			DocumentID: "did:plc:owner|sh.tangled.repo|repo1",