From 2ad517bf7e6f036453e91cc73b18fa1eca9855a6 Mon Sep 17 00:00:00 2001 From: Owais Jamil Date: Wed, 25 Mar 2026 16:54:01 -0500 Subject: [PATCH] feat: JetStream event caching and management --- docs/roadmap.md | 6 +- packages/api/internal/api/actors.go | 51 ----- packages/api/internal/api/api.go | 128 +++++++------ packages/api/internal/api/helpers.go | 151 +++++++++++++++ packages/api/internal/api/jetstream.go | 174 ++++++++++++++++++ packages/api/internal/api/jetstream_test.go | 47 +++++ packages/api/internal/api/readthrough.go | 29 --- packages/api/internal/config/config.go | 16 +- packages/api/internal/ingest/ingest_test.go | 8 + .../store/migrations/006_jetstream_events.sql | 14 ++ packages/api/internal/store/sql_store.go | 94 ++++++++++ packages/api/internal/store/store.go | 24 +++ packages/api/internal/store/store_test.go | 103 +++++++++++ 13 files changed, 699 insertions(+), 146 deletions(-) create mode 100644 packages/api/internal/api/helpers.go create mode 100644 packages/api/internal/api/jetstream.go create mode 100644 packages/api/internal/api/jetstream_test.go create mode 100644 packages/api/internal/store/migrations/006_jetstream_events.sql diff --git a/docs/roadmap.md b/docs/roadmap.md index ddef312..a18ca60 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -23,9 +23,9 @@ Highest priority. This work blocks further investment in semantic search, hybrid - [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`) diff --git a/packages/api/internal/api/actors.go b/packages/api/internal/api/actors.go index cb7a072..6c335dd 100644 --- a/packages/api/internal/api/actors.go +++ b/packages/api/internal/api/actors.go @@ -178,27 +178,6 @@ func (s *Server) actorError(w http.ResponseWriter, err error) { } } -// 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) { @@ -872,26 +851,6 @@ func (s *Server) fetchPullsAndStatuses(r *http.Request, pds, did string) ([]xrpc 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"` @@ -910,13 +869,3 @@ func (s *Server) fetchBskyProfile(r *http.Request, did string) *bskyProfileRespo } 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 -} diff --git a/packages/api/internal/api/api.go b/packages/api/internal/api/api.go index bdab831..1ec8402 100644 --- a/packages/api/internal/api/api.go +++ b/packages/api/internal/api/api.go @@ -8,7 +8,6 @@ import ( "log/slog" "net" "net/http" - "strconv" "strings" "sync" "time" @@ -48,6 +47,9 @@ func New(searchRepo *search.Repository, st store.Store, cfg *config.Config, log } // 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() @@ -63,6 +65,7 @@ func (s *Server) Handler() http.Handler { 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) @@ -120,6 +123,7 @@ func (s *Server) Run(ctx context.Context) error { 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)) @@ -407,72 +411,78 @@ func (s *Server) handleProfileSummary(w http.ResponseWriter, r *http.Request) { 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"` + } -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, + 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 (s *Server) handleNotImplemented(w http.ResponseWriter, _ *http.Request) { + writeJSON(w, http.StatusNotImplemented, errorBody("not_implemented", "this endpoint is not yet available")) } diff --git a/packages/api/internal/api/helpers.go b/packages/api/internal/api/helpers.go new file mode 100644 index 0000000..2c5a4a4 --- /dev/null +++ b/packages/api/internal/api/helpers.go @@ -0,0 +1,151 @@ +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< 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 +} diff --git a/packages/api/internal/api/jetstream.go b/packages/api/internal/api/jetstream.go new file mode 100644 index 0000000..edd4ce6 --- /dev/null +++ b/packages/api/internal/api/jetstream.go @@ -0,0 +1,174 @@ +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, "&") +} diff --git a/packages/api/internal/api/jetstream_test.go b/packages/api/internal/api/jetstream_test.go new file mode 100644 index 0000000..9b725ab --- /dev/null +++ b/packages/api/internal/api/jetstream_test.go @@ -0,0 +1,47 @@ +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) + } + }) +} diff --git a/packages/api/internal/api/readthrough.go b/packages/api/internal/api/readthrough.go index 7273cf5..0f77eab 100644 --- a/packages/api/internal/api/readthrough.go +++ b/packages/api/internal/api/readthrough.go @@ -291,32 +291,3 @@ func (s *Server) enqueueXRPCList(ctx context.Context, entries []xrpc.ListRecordE 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< 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 -} diff --git a/packages/api/internal/config/config.go b/packages/api/internal/config/config.go index 69eeb3c..194b511 100644 --- a/packages/api/internal/config/config.go +++ b/packages/api/internal/config/config.go @@ -42,8 +42,12 @@ type Config struct { 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 { @@ -85,8 +89,12 @@ func Load(opts LoadOptions) (*Config, error) { ConstellationUserAgent: envOrDefault("CONSTELLATION_USER_AGENT", "twister/1.0 (https://tangled.org/desertthunder.dev/twisted; Owais )"), 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 { diff --git a/packages/api/internal/ingest/ingest_test.go b/packages/api/internal/ingest/ingest_test.go index ee426ef..c788d03 100644 --- a/packages/api/internal/ingest/ingest_test.go +++ b/packages/api/internal/ingest/ingest_test.go @@ -147,6 +147,14 @@ func (f *fakeStore) CountPendingIndexingJobs(_ context.Context) (int64, error) { 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 } diff --git a/packages/api/internal/store/migrations/006_jetstream_events.sql b/packages/api/internal/store/migrations/006_jetstream_events.sql new file mode 100644 index 0000000..dbad8fc --- /dev/null +++ b/packages/api/internal/store/migrations/006_jetstream_events.sql @@ -0,0 +1,14 @@ +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); diff --git a/packages/api/internal/store/sql_store.go b/packages/api/internal/store/sql_store.go index c30b8e7..56194e3 100644 --- a/packages/api/internal/store/sql_store.go +++ b/packages/api/internal/store/sql_store.go @@ -469,6 +469,100 @@ func (s *SQLStore) CountPendingIndexingJobs(ctx context.Context) (int64, error) 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) } diff --git a/packages/api/internal/store/store.go b/packages/api/internal/store/store.go index 8f7bae3..dcccfba 100644 --- a/packages/api/internal/store/store.go +++ b/packages/api/internal/store/store.go @@ -56,6 +56,28 @@ type IndexingJob struct { 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 @@ -97,5 +119,7 @@ type Store interface { 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 } diff --git a/packages/api/internal/store/store_test.go b/packages/api/internal/store/store_test.go index 54aeef7..0829ecc 100644 --- a/packages/api/internal/store/store_test.go +++ b/packages/api/internal/store/store_test.go @@ -316,6 +316,109 @@ func TestIntegration(t *testing.T) { } }) + 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", -- knotmirror 0.0.1