bretton.dev
+24
-5
cmd/server/main.go
+24
-5
cmd/server/main.go
···
···+// IMPORTANT: We listen to social.coves.community.subscription (not social.coves.community.subscribe)+communityJetstreamURL = "ws://localhost:6008/subscribe?wantedCollections=social.coves.community.profile&wantedCollections=social.coves.community.subscription"+communityJetstreamConnector := jetstream.NewCommunityJetstreamConnector(communityEventConsumer, communityJetstreamURL)
+24
-12
docs/PRD_BACKLOG.md
+24
-12
docs/PRD_BACKLOG.md
···-**Problem:** Users can't control how much content they see from each community. Lexicon has `contentVisibility` (1-5 scale) but code doesn't use it.-- Update feed generation to respect visibility level (beta work, but data structure needed now)-- Lexicon: [subscription.json:28-34](../internal/atproto/lexicon/social/coves/actor/subscription.json#L28-L34) ✅ Ready-- Handler: [community/subscribe.go](../internal/api/handlers/community/subscribe.go) - Add parameter-- Service: [communities/service.go:373-376](../internal/core/communities/service.go#L373-L376) - Add to record-**Impact:** Without this, users have no way to adjust feed volume per community (key feature from DOMAIN_KNOWLEDGE.md)
···+**Problem:** Users couldn't control how much content they see from each community. Lexicon had `contentVisibility` (1-5 scale) but code didn't use it.+- Lexicon: [subscription.json](../internal/atproto/lexicon/social/coves/community/subscription.json) ✅ Updated to atProto conventions+- Handler: [community/subscribe.go](../internal/api/handlers/community/subscribe.go) ✅ Accepts contentVisibility+- Service: [communities/service.go](../internal/core/communities/service.go) ✅ Clamps and passes to PDS+- Consumer: [community_consumer.go](../internal/atproto/jetstream/community_consumer.go) ✅ Extracts and indexes+- Repository: [community_repo_subscriptions.go](../internal/db/postgres/community_repo_subscriptions.go) ✅ All queries updated+- Migration: [008_add_content_visibility_to_subscriptions.sql](../internal/db/migrations/008_add_content_visibility_to_subscriptions.sql) ✅ Schema changes+- Tests: [subscription_indexing_test.go](../tests/integration/subscription_indexing_test.go) ✅ Comprehensive coverage+**Documentation:** See [IMPLEMENTATION_SUBSCRIPTION_INDEXING.md](../docs/IMPLEMENTATION_SUBSCRIPTION_INDEXING.md) for full details+**Impact:** ✅ Users can now adjust feed volume per community (key feature from DOMAIN_KNOWLEDGE.md enabled)
+30
-23
docs/PRD_COMMUNITIES.md
+30
-23
docs/PRD_COMMUNITIES.md
······-- [ ] **Subscription Visibility Level (1-5 Scale):** Implement feed slider from DOMAIN_KNOWLEDGE.md-- Lexicon: ✅ Ready ([subscription.json:28-34](internal/atproto/lexicon/social/coves/actor/subscription.json))···-2. Subscribe to local Jetstream: `ws://localhost:6008/subscribe?wantedCollections=social.coves.community.subscribe`-3. Fix unsubscribe handler - should handle `delete` operation on `social.coves.community.subscribe`, NOT a separate collection-4. Remove incorrect `social.coves.community.unsubscribe` case ([community_consumer.go:40](internal/atproto/jetstream/community_consumer.go#L40))-- Consumer: [internal/atproto/jetstream/community_consumer.go](internal/atproto/jetstream/community_consumer.go) (exists, needs fixes)
·········+1. ✅ Fixed critical collection name bug (`social.coves.actor.subscription` → `social.coves.community.subscription`)+2. ✅ Implemented ContentVisibility (1-5 slider) across all layers (handler, service, consumer, repository)+3. ✅ Production Jetstream consumer now running ([cmd/server/main.go:220-243](cmd/server/main.go#L220-L243))+- Implementation Doc: [docs/IMPLEMENTATION_SUBSCRIPTION_INDEXING.md](docs/IMPLEMENTATION_SUBSCRIPTION_INDEXING.md)+- Lexicon: [internal/atproto/lexicon/social/coves/community/subscription.json](internal/atproto/lexicon/social/coves/community/subscription.json)+- Consumer: [internal/atproto/jetstream/community_consumer.go](internal/atproto/jetstream/community_consumer.go)+- Connector: [internal/atproto/jetstream/community_jetstream_connector.go](internal/atproto/jetstream/community_jetstream_connector.go)+- Migration: [internal/db/migrations/008_add_content_visibility_to_subscriptions.sql](internal/db/migrations/008_add_content_visibility_to_subscriptions.sql)+- Tests: [tests/integration/subscription_indexing_test.go](tests/integration/subscription_indexing_test.go)
+4
-2
internal/api/handlers/community/subscribe.go
+4
-2
internal/api/handlers/community/subscribe.go
······// Extract authenticated user DID and access token from request context (injected by auth middleware)···-subscription, err := h.service.SubscribeToCommunity(r.Context(), userDID, userAccessToken, req.Community)
······// Extract authenticated user DID and access token from request context (injected by auth middleware)···+subscription, err := h.service.SubscribeToCommunity(r.Context(), userDID, userAccessToken, req.Community, req.ContentVisibility)
+112
-34
internal/atproto/jetstream/community_consumer.go
+112
-34
internal/atproto/jetstream/community_consumer.go
······func (c *CommunityEventConsumer) handleSubscription(ctx context.Context, userDID string, commit *CommitEvent) error {-func (c *CommunityEventConsumer) handleUnsubscribe(ctx context.Context, userDID string, commit *CommitEvent) error {···
······func (c *CommunityEventConsumer) handleSubscription(ctx context.Context, userDID string, commit *CommitEvent) error {+func (c *CommunityEventConsumer) createSubscription(ctx context.Context, userDID string, commit *CommitEvent) error {+// IMPORTANT: Collection is social.coves.community.subscription (record type), not the XRPC endpoint+func (c *CommunityEventConsumer) deleteSubscription(ctx context.Context, userDID string, commit *CommitEvent) error {···
+136
internal/atproto/jetstream/community_jetstream_connector.go
+136
internal/atproto/jetstream/community_jetstream_connector.go
···
···+// NewCommunityJetstreamConnector creates a new Jetstream WebSocket connector for community events+func NewCommunityJetstreamConnector(consumer *CommunityEventConsumer, wsURL string) *CommunityJetstreamConnector {
+4
-5
internal/atproto/jetstream/user_consumer.go
+4
-5
internal/atproto/jetstream/user_consumer.go
·········
·········
+7
-6
internal/core/communities/community.go
+7
-6
internal/core/communities/community.go
···
···+ContentVisibility int `json:"contentVisibility" db:"content_visibility"` // Feed slider: 1-5 (1=best content only, 5=all content)
+2
-1
internal/core/communities/interfaces.go
+2
-1
internal/core/communities/interfaces.go
···UnsubscribeWithCount(ctx context.Context, userDID, communityDID string) error // Atomic: unsubscribe + decrement countListSubscriptions(ctx context.Context, userDID string, limit, offset int) ([]*Subscription, error)ListSubscribers(ctx context.Context, communityDID string, limit, offset int) ([]*Subscription, error)···-SubscribeToCommunity(ctx context.Context, userDID, userAccessToken, communityIdentifier string) (*Subscription, error)UnsubscribeFromCommunity(ctx context.Context, userDID, userAccessToken, communityIdentifier string) errorGetUserSubscriptions(ctx context.Context, userDID string, limit, offset int) ([]*Subscription, error)GetCommunitySubscribers(ctx context.Context, communityIdentifier string, limit, offset int) ([]*Subscription, error)
···UnsubscribeWithCount(ctx context.Context, userDID, communityDID string) error // Atomic: unsubscribe + decrement count+GetSubscriptionByURI(ctx context.Context, recordURI string) (*Subscription, error) // For Jetstream delete operationsListSubscriptions(ctx context.Context, userDID string, limit, offset int) ([]*Subscription, error)ListSubscribers(ctx context.Context, communityDID string, limit, offset int) ([]*Subscription, error)···+SubscribeToCommunity(ctx context.Context, userDID, userAccessToken, communityIdentifier string, contentVisibility int) (*Subscription, error)UnsubscribeFromCommunity(ctx context.Context, userDID, userAccessToken, communityIdentifier string) errorGetUserSubscriptions(ctx context.Context, userDID string, limit, offset int) ([]*Subscription, error)GetCommunitySubscribers(ctx context.Context, communityIdentifier string, limit, offset int) ([]*Subscription, error)
+23
-10
internal/core/communities/service.go
+23
-10
internal/core/communities/service.go
···-func (s *communityService) SubscribeToCommunity(ctx context.Context, userDID, userAccessToken, communityIdentifier string) (*Subscription, error) {···-recordURI, recordCID, err := s.createRecordOnPDSAs(ctx, userDID, "social.coves.community.subscribe", "", subRecord, userAccessToken)···-if err := s.deleteRecordOnPDSAs(ctx, userDID, "social.coves.community.subscribe", rkey, userAccessToken); err != nil {
···+func (s *communityService) SubscribeToCommunity(ctx context.Context, userDID, userAccessToken, communityIdentifier string, contentVisibility int) (*Subscription, error) {···+// CRITICAL: Collection is social.coves.community.subscription (RECORD TYPE), not social.coves.community.subscribe (XRPC procedure)+// This record will be created in the USER's repository: at://user_did/social.coves.community.subscription/{tid}+recordURI, recordCID, err := s.createRecordOnPDSAs(ctx, userDID, "social.coves.community.subscription", "", subRecord, userAccessToken)···+// CRITICAL: Delete from social.coves.community.subscription (RECORD TYPE), not social.coves.community.unsubscribe+if err := s.deleteRecordOnPDSAs(ctx, userDID, "social.coves.community.subscription", rkey, userAccessToken); err != nil {
+23
internal/db/migrations/008_add_content_visibility_to_subscriptions.sql
+23
internal/db/migrations/008_add_content_visibility_to_subscriptions.sql
···
···+CREATE INDEX idx_subscriptions_user_visibility ON community_subscriptions(user_did, content_visibility);+COMMENT ON COLUMN community_subscriptions.content_visibility IS 'Feed slider: 1=only best content, 5=all content (see social.coves.community.subscription lexicon)';
+50
-11
internal/db/postgres/community_repo_subscriptions.go
+50
-11
internal/db/postgres/community_repo_subscriptions.go
···func (r *postgresCommunityRepo) Subscribe(ctx context.Context, subscription *communities.Subscription) (*communities.Subscription, error) {-INSERT INTO community_subscriptions (user_did, community_did, subscribed_at, record_uri, record_cid)······-INSERT INTO community_subscriptions (user_did, community_did, subscribed_at, record_uri, record_cid)···-query = `SELECT id, subscribed_at FROM community_subscriptions WHERE user_did = $1 AND community_did = $2`-err = tx.QueryRowContext(ctx, query, subscription.UserDID, subscription.CommunityDID).Scan(&subscription.ID, &subscription.SubscribedAt)···func (r *postgresCommunityRepo) GetSubscription(ctx context.Context, userDID, communityDID string) (*communities.Subscription, error) {······func (r *postgresCommunityRepo) ListSubscriptions(ctx context.Context, userDID string, limit, offset int) ([]*communities.Subscription, error) {······func (r *postgresCommunityRepo) ListSubscribers(ctx context.Context, communityDID string, limit, offset int) ([]*communities.Subscription, error) {···
···func (r *postgresCommunityRepo) Subscribe(ctx context.Context, subscription *communities.Subscription) (*communities.Subscription, error) {+INSERT INTO community_subscriptions (user_did, community_did, subscribed_at, record_uri, record_cid, content_visibility)······+INSERT INTO community_subscriptions (user_did, community_did, subscribed_at, record_uri, record_cid, content_visibility)···+query = `SELECT id, subscribed_at, content_visibility FROM community_subscriptions WHERE user_did = $1 AND community_did = $2`+err = tx.QueryRowContext(ctx, query, subscription.UserDID, subscription.CommunityDID).Scan(&subscription.ID, &subscription.SubscribedAt, &subscription.ContentVisibility)···func (r *postgresCommunityRepo) GetSubscription(ctx context.Context, userDID, communityDID string) (*communities.Subscription, error) {······+func (r *postgresCommunityRepo) GetSubscriptionByURI(ctx context.Context, recordURI string) (*communities.Subscription, error) {func (r *postgresCommunityRepo) ListSubscriptions(ctx context.Context, userDID string, limit, offset int) ([]*communities.Subscription, error) {······func (r *postgresCommunityRepo) ListSubscribers(ctx context.Context, communityDID string, limit, offset int) ([]*communities.Subscription, error) {···
+5
-2
tests/integration/community_consumer_test.go
+5
-2
tests/integration/community_consumer_test.go
······
······
+178
-18
tests/integration/community_e2e_test.go
+178
-18
tests/integration/community_e2e_test.go
······pdsResp, pdsErr := http.Get(fmt.Sprintf("%s/xrpc/com.atproto.repo.getRecord?repo=%s&collection=%s&rkey=%s",······-t.Errorf("Community DID mismatch: expected %s, got %v", community.DID, pdsRecord.Value["community"])-subscription, err := communityService.SubscribeToCommunity(ctx, instanceDID, accessToken, community.DID)······pdsResp, pdsErr := http.Get(fmt.Sprintf("%s/xrpc/com.atproto.repo.getRecord?repo=%s&collection=%s&rkey=%s",···
······pdsResp, pdsErr := http.Get(fmt.Sprintf("%s/xrpc/com.atproto.repo.getRecord?repo=%s&collection=%s&rkey=%s",···+t.Fatalf("Subscription record not found on PDS: status %d, body: %s", pdsResp.StatusCode, string(body))···+t.Errorf("Community DID mismatch: expected %s, got %v", community.DID, pdsRecord.Value["subject"])+t.Errorf("ContentVisibility not indexed correctly: expected 5, got %d", indexedSub.ContentVisibility)+subscription, err := communityService.SubscribeToCommunity(ctx, instanceDID, accessToken, community.DID, 3)······pdsResp, pdsErr := http.Get(fmt.Sprintf("%s/xrpc/com.atproto.repo.getRecord?repo=%s&collection=%s&rkey=%s",···
+76
-72
tests/integration/community_repo_test.go
+76
-72
tests/integration/community_repo_test.go
············
············
+499
tests/integration/subscription_indexing_test.go
+499
tests/integration/subscription_indexing_test.go
···
···+t.Errorf("Input %.0f: expected %d, got %d", tc.input, tc.expected, subscription.ContentVisibility)+t.Run("idempotent delete: deleting non-existent subscription doesn't fail", func(t *testing.T) {+// TestSubscriptionIndexing_SubscriberCount tests that subscriber counts are updated atomically+func createTestCommunity(t *testing.T, repo communities.Repository, name, did string) *communities.Community {
-6
tests/lexicon-test-data/actor/subscription-invalid-visibility.json
-6
tests/lexicon-test-data/actor/subscription-invalid-visibility.json
-6
tests/lexicon-test-data/actor/subscription-valid.json
-6
tests/lexicon-test-data/actor/subscription-valid.json
+6
tests/lexicon-test-data/community/subscription-invalid-visibility.json
+6
tests/lexicon-test-data/community/subscription-invalid-visibility.json
+6
tests/lexicon-test-data/community/subscription-valid.json
+6
tests/lexicon-test-data/community/subscription-valid.json
+4
tests/unit/community_service_test.go
+4
tests/unit/community_service_test.go
···+func (m *mockCommunityRepo) GetSubscriptionByURI(ctx context.Context, recordURI string) (*communities.Subscription, error) {func (m *mockCommunityRepo) ListSubscriptions(ctx context.Context, userDID string, limit, offset int) ([]*communities.Subscription, error) {