feat: constellation integration

docs/roadmap.md

···
 
 Add a Constellation client to the Go API for enriching search results with social signals.
 
-- [ ] Constellation XRPC client (`internal/constellation/`) with `getBacklinksCount` and `getBacklinks`
-- [ ] User-agent header with project name and contact
-- [ ] Enrich search results with star counts from Constellation
-- [ ] Profile summary endpoint (`GET /profiles/{did}/summary`) with follower/following counts from Constellation
-- [ ] Cache Constellation responses with short TTL (star/follower counts change infrequently)
+- [x] Constellation XRPC client (`internal/constellation/`) with `getBacklinksCount` and `getBacklinks`
+- [x] User-agent header with project name and contact
+- [x] Enrich search results with star counts from Constellation
+- [x] Profile summary endpoint (`GET /profiles/{did}/summary`) with follower/following counts from Constellation
+- [x] Cache Constellation responses with short TTL (star/follower counts change infrequently)
 
 ## API: Semantic Search Pipeline
 

packages/api/internal/api/api.go

···
 	"net/http"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"tangled.org/desertthunder.dev/twister/internal/config"
+	"tangled.org/desertthunder.dev/twister/internal/constellation"
 	"tangled.org/desertthunder.dev/twister/internal/reindex"
 	"tangled.org/desertthunder.dev/twister/internal/search"
 	"tangled.org/desertthunder.dev/twister/internal/store"
···
 
 // Server is the HTTP search API server.
 type Server struct {
-	search *search.Repository
-	store  store.Store
-	cfg    *config.Config
-	log    *slog.Logger
+	search        *search.Repository
+	store         store.Store
+	cfg           *config.Config
+	log           *slog.Logger
+	constellation *constellation.Client
 }
 
 // New creates a new API server.
-func New(searchRepo *search.Repository, st store.Store, cfg *config.Config, log *slog.Logger) *Server {
+func New(searchRepo *search.Repository, st store.Store, cfg *config.Config, log *slog.Logger, constellation *constellation.Client) *Server {
 	return &Server{
-		search: searchRepo,
-		store:  st,
-		cfg:    cfg,
-		log:    log,
+		search:        searchRepo,
+		store:         st,
+		cfg:           cfg,
+		log:           log,
+		constellation: constellation,
 	}
 }
 
···
 	mux.HandleFunc("GET /search/hybrid", s.handleNotImplemented)
 
 	mux.HandleFunc("GET /documents/{id}", s.handleGetDocument)
+	mux.HandleFunc("GET /profiles/{did}/summary", s.handleProfileSummary)
 
 	if s.cfg.EnableAdminEndpoints {
 		mux.HandleFunc("POST /admin/reindex", s.handleAdminReindex)
···
 		return
 	}
 
+	if s.constellation != nil {
+		s.enrichStarCounts(r.Context(), resp.Results)
+	}
+
 	writeJSON(w, http.StatusOK, resp)
 }
 
···
 		"updated": result.Updated,
 		"errors":  result.Errors,
 	})
+}
+
+// enrichStarCounts fetches star counts from Constellation for repo results in parallel.
+// It is best-effort: failures are logged and results are returned without star counts.
+//
+// Uses a short deadline so enrichment doesn't stall the response.
+func (s *Server) enrichStarCounts(ctx context.Context, results []search.Result) {
+	ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
+	defer cancel()
+
+	var wg sync.WaitGroup
+	for i := range results {
+		if results[i].RecordType != "repo" || results[i].ATURI == "" {
+			continue
+		}
+		wg.Add(1)
+		go func(i int) {
+			defer wg.Done()
+			n, err := s.constellation.GetBacklinksCount(ctx, constellation.BacklinksParams{
+				Subject: results[i].ATURI,
+				Source:  constellation.SourceStarURI,
+			})
+			if err != nil {
+				s.log.Debug("constellation star count failed", slog.String("at_uri", results[i].ATURI), slog.String("error", err.Error()))
+				return
+			}
+			results[i].StarCount = &n
+		}(i)
+	}
+	wg.Wait()
+}
+
+// handleProfileSummary returns follower count and other social signals for a DID.
+func (s *Server) handleProfileSummary(w http.ResponseWriter, r *http.Request) {
+	did := r.PathValue("did")
+	if did == "" {
+		writeJSON(w, http.StatusBadRequest, errorBody("invalid_parameter", "did is required"))
+		return
+	}
+
+	type summaryResponse struct {
+		DID           string `json:"did"`
+		FollowerCount int    `json:"follower_count"`
+	}
+
+	summary := summaryResponse{DID: did}
+
+	if s.constellation != nil {
+		n, err := s.constellation.GetBacklinksCount(r.Context(), constellation.BacklinksParams{
+			Subject: did,
+			Source:  constellation.SourceFollowDID,
+		})
+		if err != nil {
+			s.log.Debug("constellation follower count failed", slog.String("did", did), slog.String("error", err.Error()))
+		} else {
+			summary.FollowerCount = n
+		}
+	}
+
+	writeJSON(w, http.StatusOK, summary)
 }
 
 func (s *Server) handleNotImplemented(w http.ResponseWriter, _ *http.Request) {

packages/api/internal/config/config.go

···
 	PLCDirectoryURL        string
 	IdentityServiceURL     string
 	XRPCTimeout            time.Duration
+	ConstellationURL       string
+	ConstellationUserAgent string
+	ConstellationTimeout   time.Duration
+	ConstellationCacheTTL  time.Duration
 }
 
 type LoadOptions struct {
···
 		PLCDirectoryURL:        envOrDefault("PLC_DIRECTORY_URL", "https://plc.directory"),
 		IdentityServiceURL:     envOrDefault("IDENTITY_SERVICE_URL", "https://public.api.bsky.app"),
 		XRPCTimeout:            envDuration("XRPC_TIMEOUT", 15*time.Second),
+		ConstellationURL:       envOrDefault("CONSTELLATION_URL", "https://constellation.microcosm.blue"),
+		ConstellationUserAgent: envOrDefault("CONSTELLATION_USER_AGENT", "twister/1.0 (https://tangled.sh; Owais <[e-mail]>)"),
+		ConstellationTimeout:   envDuration("CONSTELLATION_TIMEOUT", 10*time.Second),
+		ConstellationCacheTTL:  envDuration("CONSTELLATION_CACHE_TTL", 5*time.Minute),
 	}
 
 	if opts.Local {

packages/api/internal/constellation/client.go

···
+// Package constellation provides a client for the Constellation backlink API.
+// Constellation is a public AT Protocol backlink index at https://constellation.microcosm.blue.
+// It answers "who linked to this?" across the network, providing social signal counts
+// (stars, followers, reactions) without requiring us to maintain our own counters.
+package constellation
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"time"
+)
+
+const (
+	DefaultBaseURL  string        = "https://constellation.microcosm.blue"
+	defaultTimeout  time.Duration = 10 * time.Second
+	defaultCacheTTL time.Duration = 5 * time.Minute
+)
+
+// Source constants for common Tangled collections.
+const (
+	SourceStarURI     string = "sh.tangled.feed.star:subject.uri"
+	SourceFollowDID   string = "sh.tangled.graph.follow:subject"
+	SourceReactionURI string = "sh.tangled.feed.reaction:subject.uri"
+)
+
+// Client is an HTTP client for the Constellation backlink API.
+type Client struct {
+	http       *http.Client
+	baseURL    string
+	userAgent  string
+	countCache *ttlCache[int]
+}
+
+// Option configures a Client.
+type Option func(*Client)
+
+// WithBaseURL overrides the Constellation base URL.
+func WithBaseURL(u string) Option {
+	return func(c *Client) { c.baseURL = u }
+}
+
+// WithUserAgent sets the User-Agent header sent with every request.
+// Constellation requires a user-agent identifying the project and a contact.
+func WithUserAgent(ua string) Option {
+	return func(c *Client) { c.userAgent = ua }
+}
+
+// WithTimeout sets the HTTP request timeout.
+func WithTimeout(d time.Duration) Option {
+	return func(c *Client) { c.http.Timeout = d }
+}
+
+// WithCacheTTL sets the TTL for cached count responses.
+func WithCacheTTL(d time.Duration) Option {
+	return func(c *Client) { c.countCache = newTTLCache[int](d) }
+}
+
+// NewClient creates a Constellation client with sensible defaults.
+func NewClient(opts ...Option) *Client {
+	c := &Client{
+		http:       &http.Client{Timeout: defaultTimeout},
+		baseURL:    DefaultBaseURL,
+		userAgent:  "twister/1.0",
+		countCache: newTTLCache[int](defaultCacheTTL),
+	}
+	for _, o := range opts {
+		o(c)
+	}
+	return c
+}
+
+// BacklinksParams holds query parameters for backlink endpoints.
+type BacklinksParams struct {
+	// Subject is the target AT-URI or DID being linked to (required).
+	Subject string
+	// Source is the collection path, e.g. "sh.tangled.feed.star:subject.uri" (required).
+	Source string
+	// DID optionally filters results to a specific actor (repeatable in the raw API; here single-value).
+	DID string
+	// Limit is the max results to return (default 16, max 100).
+	Limit int
+	// Reverse reverses the ordering.
+	Reverse bool
+}
+
+// BacklinkRecord is one entry returned by GetBacklinks.
+type BacklinkRecord struct {
+	URI       string `json:"uri"`
+	CID       string `json:"cid"`
+	ActorDID  string `json:"actorDid"`
+	CreatedAt string `json:"createdAt"`
+}
+
+// BacklinksResponse is returned by GetBacklinks.
+type BacklinksResponse struct {
+	Records []BacklinkRecord `json:"records"`
+	Cursor  string           `json:"cursor,omitempty"`
+}
+
+// GetBacklinksCount returns the count of records linking to the given subject.
+// Results are cached with the configured TTL. Errors are returned without caching.
+func (c *Client) GetBacklinksCount(ctx context.Context, p BacklinksParams) (int, error) {
+	cacheKey := "count\x00" + p.Subject + "\x00" + p.Source
+	if n, ok := c.countCache.Get(cacheKey); ok {
+		return n, nil
+	}
+
+	params := url.Values{}
+	params.Set("subject", p.Subject)
+	params.Set("source", p.Source)
+
+	var resp struct {
+		Count int `json:"count"`
+	}
+	if err := c.get(ctx, "blue.microcosm.links.getBacklinksCount", params, &resp); err != nil {
+		return 0, err
+	}
+
+	c.countCache.Set(cacheKey, resp.Count)
+	return resp.Count, nil
+}
+
+// GetBacklinks returns records linking to the given subject.
+// Results are not cached because paginated lists change frequently.
+func (c *Client) GetBacklinks(ctx context.Context, p BacklinksParams) (*BacklinksResponse, error) {
+	params := url.Values{}
+	params.Set("subject", p.Subject)
+	params.Set("source", p.Source)
+	if p.DID != "" {
+		params.Set("did", p.DID)
+	}
+	if p.Limit > 0 {
+		params.Set("limit", fmt.Sprintf("%d", p.Limit))
+	}
+	if p.Reverse {
+		params.Set("reverse", "true")
+	}
+
+	var resp BacklinksResponse
+	if err := c.get(ctx, "blue.microcosm.links.getBacklinks", params, &resp); err != nil {
+		return nil, err
+	}
+	return &resp, nil
+}
+
+func (c *Client) get(ctx context.Context, method string, params url.Values, out any) error {
+	u := c.baseURL + "/xrpc/" + method
+	if len(params) > 0 {
+		u += "?" + params.Encode()
+	}
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil)
+	if err != nil {
+		return fmt.Errorf("constellation: build request: %w", err)
+	}
+	if c.userAgent != "" {
+		req.Header.Set("User-Agent", c.userAgent)
+	}
+
+	resp, err := c.http.Do(req)
+	if err != nil {
+		return fmt.Errorf("constellation: request %s: %w", method, err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		body, _ := io.ReadAll(resp.Body)
+		var errResp struct {
+			Message string `json:"message"`
+		}
+		if json.Unmarshal(body, &errResp) == nil && errResp.Message != "" {
+			return fmt.Errorf("constellation: %s: status %d: %s", method, resp.StatusCode, errResp.Message)
+		}
+		return fmt.Errorf("constellation: %s: status %d", method, resp.StatusCode)
+	}
+
+	if err := json.NewDecoder(resp.Body).Decode(out); err != nil {
+		return fmt.Errorf("constellation: %s: decode response: %w", method, err)
+	}
+	return nil
+}

packages/api/internal/constellation/ttl.go

···
+package constellation
+
+import (
+	"sync"
+	"time"
+)
+
+type cacheEntry[T any] struct {
+	value     T
+	expiresAt time.Time
+}
+
+type ttlCache[T any] struct {
+	mu      sync.RWMutex
+	entries map[string]cacheEntry[T]
+	ttl     time.Duration
+}
+
+func newTTLCache[T any](ttl time.Duration) *ttlCache[T] {
+	return &ttlCache[T]{
+		entries: make(map[string]cacheEntry[T]),
+		ttl:     ttl,
+	}
+}
+
+func (c *ttlCache[T]) Get(key string) (T, bool) {
+	c.mu.RLock()
+	entry, ok := c.entries[key]
+	c.mu.RUnlock()
+	if !ok {
+		var zero T
+		return zero, false
+	}
+	if time.Now().After(entry.expiresAt) {
+		c.mu.Lock()
+		delete(c.entries, key)
+		c.mu.Unlock()
+		var zero T
+		return zero, false
+	}
+	return entry.value, true
+}
+
+func (c *ttlCache[T]) Set(key string, value T) {
+	c.mu.Lock()
+	c.entries[key] = cacheEntry[T]{
+		value:     value,
+		expiresAt: time.Now().Add(c.ttl),
+	}
+	c.mu.Unlock()
+}

packages/api/internal/enrich/enrich.go

···
 
 // Runner performs the enrichment operation.
 type Runner struct {
-	store  store.Store
-	xrpc   *xrpc.Client
-	log    *slog.Logger
+	store store.Store
+	xrpc  *xrpc.Client
+	log   *slog.Logger
 }
 
 // New creates a Runner.
···
 func (r *Runner) enrichDoc(ctx context.Context, doc *store.Document) bool {
 	changed := false
 
-	// Resolve author handle
 	if doc.AuthorHandle == "" && doc.DID != "" {
 		handle, err := r.store.GetIdentityHandle(ctx, doc.DID)
 		if err == nil && handle != "" {
···
 		}
 	}
 
-	// Resolve repo name for repo-scoped records
 	if doc.RepoDID != "" && doc.RepoName == "" {
-		// Try to find the repo name from an existing repo document in the store
 		repoName := r.findRepoNameFromStore(ctx, doc.RepoDID)
 		if repoName != "" {
 			doc.RepoName = repoName
···
 		}
 	}
 
-	// Resolve repo owner handle for WebURL
 	ownerHandle := doc.AuthorHandle
 	if doc.RepoDID != "" && doc.RepoDID != doc.DID {
 		repoOwnerHandle, err := r.store.GetIdentityHandle(ctx, doc.RepoDID)
···
 		}
 	}
 
-	// Build WebURL
 	if doc.WebURL == "" {
 		webURL := xrpc.BuildWebURL(ownerHandle, doc.RepoName, doc.RecordType, doc.RKey)
 		if webURL != "" {
···
 	if err != nil || len(docs) == 0 {
 		return ""
 	}
-	// Return the first repo's title (which is the repo name)
 	for _, d := range docs {
 		if d.Title != "" {
 			return d.Title

packages/api/internal/ingest/ingest.go

···
 		return
 	}
 
-	// Resolve repo name if RepoDID is set but RepoName is empty
 	if doc.RepoDID != "" && doc.RepoName == "" {
-		// Extract the repo rkey from the RepoDID — the repo AT-URI typically encodes
-		// the rkey as the last segment. We try to look up the repo record.
-		// The RepoDID in the document refers to the repo owner's DID. The repo rkey
-		// can be extracted from the document's AT-URI for repo-scoped records.
 		repoRKey := extractRepoRKey(doc.ATURI, doc.Collection)
 		if repoRKey != "" {
 			name, err := r.xrpcClient.ResolveRepoName(ctx, doc.RepoDID, repoRKey)
···
 		}
 	}
 
-	// Resolve author handle if empty
 	if doc.AuthorHandle == "" && doc.DID != "" {
 		info, err := r.xrpcClient.ResolveIdentity(ctx, doc.DID)
 		if err != nil {
···
 		}
 	}
 
-	// Build WebURL if we have enough data
 	if doc.WebURL == "" {
 		ownerHandle := doc.AuthorHandle
 		if doc.RepoDID != "" && doc.RepoDID != doc.DID {
-			// Repo owner may differ from author — try to resolve repo owner handle
 			repoOwnerHandle, err := r.store.GetIdentityHandle(ctx, doc.RepoDID)
 			if err == nil && repoOwnerHandle != "" {
 				ownerHandle = repoOwnerHandle
···
 // at://did/collection/rkey but the repo is identified by RepoDID. We look
 // for a stored repo document, or try common rkey patterns.
 func extractRepoRKey(atURI, collection string) string {
-	// For repo records themselves, the rkey IS the repo rkey
 	if collection == "sh.tangled.repo" {
 		parts := strings.SplitN(atURI, "/", 5)
 		if len(parts) >= 5 {
 			return parts[4]
 		}
 	}
-	// For sub-collections like sh.tangled.repo.issue, the AT-URI contains the
-	// issue rkey, not the repo rkey. We can't derive the repo rkey from the URI.
-	// This will be resolved by the enrich command for existing documents.
 	return ""
 }
 

packages/api/internal/search/search.go

···
 	WebURL          string   `json:"web_url,omitempty"`
 	Score           float64  `json:"score"`
 	MatchedBy       []string `json:"matched_by"`
+	StarCount       *int     `json:"star_count,omitempty"`
 	CreatedAt       string   `json:"created_at,omitempty"`
 	UpdatedAt       string   `json:"updated_at,omitempty"`
 }

packages/api/internal/xrpc/did_test.go

···
 	}))
 	defer srv.Close()
 
-	// For did:web, we need to override how the URL is constructed.
-	// We'll use a did:plc: test instead since did:web requires DNS resolution.
-	// The PLC test above validates the parsing logic. Let's test ResolveIdentity instead.
 	t.Skip("did:web requires DNS; tested via PLC path")
 }
 
···
 
 	c := NewClient(WithPLCDirectory(srv.URL))
 
-	// First call: cache miss
 	_, err := c.ResolveDIDDoc(context.Background(), "did:plc:cached")
 	if err != nil {
 		t.Fatal(err)
···
 		t.Fatalf("expected 1 call, got %d", calls)
 	}
 
-	// Second call: cache hit
 	_, err = c.ResolveDIDDoc(context.Background(), "did:plc:cached")
 	if err != nil {
 		t.Fatal(err)
···
 		t.Fatalf("expected 1 call (cached), got %d", calls)
 	}
 
-	// Invalidate
 	c.didCache.Invalidate("did:plc:cached")
 	_, err = c.ResolveDIDDoc(context.Background(), "did:plc:cached")
 	if err != nil {

packages/api/main.go

···
 	"tangled.org/desertthunder.dev/twister/internal/api"
 	"tangled.org/desertthunder.dev/twister/internal/backfill"
 	"tangled.org/desertthunder.dev/twister/internal/config"
+	"tangled.org/desertthunder.dev/twister/internal/constellation"
 	"tangled.org/desertthunder.dev/twister/internal/enrich"
 	"tangled.org/desertthunder.dev/twister/internal/ingest"
 	"tangled.org/desertthunder.dev/twister/internal/normalize"
···
 
 			st := store.New(db)
 			searchRepo := search.NewRepository(db)
-			srv := api.New(searchRepo, st, cfg, log)
+
+			constellationClient := constellation.NewClient(
+				constellation.WithBaseURL(cfg.ConstellationURL),
+				constellation.WithUserAgent(cfg.ConstellationUserAgent),
+				constellation.WithTimeout(cfg.ConstellationTimeout),
+				constellation.WithCacheTTL(cfg.ConstellationCacheTTL),
+			)
+			log.Info("constellation client configured", slog.String("url", cfg.ConstellationURL))
+
+			srv := api.New(searchRepo, st, cfg, log, constellationClient)
 
 			ctx, cancel := baseContext()
 			defer cancel()