bretton.dev
-1
CLAUDE.md
-1
CLAUDE.md
+37
-2
cmd/server/main.go
+37
-2
cmd/server/main.go
············
······+postService := posts.NewPostService(postRepo, communityService, aggregatorService, defaultPDS)···+aggregatorJetstreamURL = "ws://localhost:6008/subscribe?wantedCollections=social.coves.aggregator.service&wantedCollections=social.coves.aggregator.authorization"+aggregatorJetstreamConnector := jetstream.NewAggregatorJetstreamConnector(aggregatorEventConsumer, aggregatorJetstreamURL)···
+224
-1278
docs/aggregators/PRD_AGGREGATORS.md
+224
-1278
docs/aggregators/PRD_AGGREGATORS.md
···-Coves Aggregators are autonomous services that automatically post content to communities. Each aggregator is identified by its own DID and operates as a specialized actor within the atProto ecosystem. This system enables communities to have automated content feeds (RSS, sports results, TV/movie discussion threads, Bluesky mirrors, etc.) while maintaining full community control over which aggregators can post and what content they can create.**Key Differentiator:** Unlike other platforms where users manually aggregate content, Coves communities can enable automated aggregators to handle routine posting tasks, creating a more dynamic and up-to-date community experience.-This record declares the existence of an aggregator service and provides metadata for discovery.-"description": "JSON Schema describing config options for this aggregator. Communities use this to know what configuration fields are available."-"description": "Automatically posts breaking news from configured RSS feeds with LLM-powered deduplication",-This record grants an aggregator permission to post to a community and contains aggregator-specific configuration.-"description": "Authorization for an aggregator to post to a community with specific configuration",-"description": "Whether this aggregator is currently active. Can be toggled without deleting the record."-"description": "Aggregator-specific configuration. Must conform to the aggregator's configSchema."···-CREATE INDEX idx_aggregator_auth_agg_did ON aggregator_authorizations(aggregator_did) WHERE enabled = true;-CREATE INDEX idx_aggregator_auth_comm_did ON aggregator_authorizations(community_did) WHERE enabled = true;-CREATE INDEX idx_aggregator_posts_agg_did_created ON aggregator_posts(aggregator_did, created_at DESC);-CREATE INDEX idx_aggregator_posts_comm_did_created ON aggregator_posts(community_did, created_at DESC);-CREATE INDEX idx_aggregator_posts_rate_limit ON aggregator_posts(aggregator_did, community_did, created_at DESC);-ListAuthorizationsForAggregator(ctx context.Context, aggDID string, enabledOnly bool) ([]*Authorization, error)-EnableAggregator(ctx context.Context, commDID, aggDID string, config map[string]interface{}) (*Authorization, error)-UpdateAggregatorConfig(ctx context.Context, commDID, aggDID string, config map[string]interface{}) error-ListCommunityAggregators(ctx context.Context, commDID string, enabledOnly bool) ([]*AggregatorInfo, error)-if err := aggregatorService.ValidateAggregatorPost(ctx, actorDID, input.Community); err != nil {-func (a *BaseAggregator) CreatePost(ctx context.Context, communityDID string, post Post) error {-func (a *BaseAggregator) GetAuthorizedCommunities(ctx context.Context) ([]*CommunityAuth, error) {-**Decision:** Aggregators use existing post creation endpoint, not a separate `social.coves.aggregator.post.create`-- Coves Posts Implementation: [IMPLEMENTATION_POST_CREATION.md](IMPLEMENTATION_POST_CREATION.md)
···+Coves Aggregators are autonomous services that automatically post content to communities. Each aggregator is identified by its own DID and operates as a specialized actor within the atProto ecosystem. This enables communities to have automated content feeds (RSS, sports results, TV/movie discussion threads, Bluesky mirrors, etc.) while maintaining full community control.**Key Differentiator:** Unlike other platforms where users manually aggregate content, Coves communities can enable automated aggregators to handle routine posting tasks, creating a more dynamic and up-to-date community experience.+Aggregators follow established atProto patterns for autonomous services (Feed Generators + Labelers model):···+- `idx_aggregator_auth_lookup` - Fast (aggregator_did, community_did, enabled) lookups for post creation+- Add branching logic in post handler: if aggregator, check authorization; else check membership+Monitors specific users/hashtags on Bluesky, creates posts in community with original author metadata.
+3
go.mod
+3
go.mod
···
···
+6
go.sum
+6
go.sum
···github.com/warpfork/go-wish v0.0.0-20220906213052-39a1cc7a02d0/go.mod h1:x6AKhvSSexNrVSrViXSHUEbICjmGXhtgABaHIySUSGw=github.com/whyrusleeping/cbor-gen v0.2.1-0.20241030202151-b7a6831be65e h1:28X54ciEwwUxyHn9yrZfl5ojgF4CBNLWX7LR0rvBkf4=github.com/whyrusleeping/cbor-gen v0.2.1-0.20241030202151-b7a6831be65e/go.mod h1:pM99HXyEbSQHcosHc0iW7YFmwnscr+t9Te4ibko05so=
···github.com/warpfork/go-wish v0.0.0-20220906213052-39a1cc7a02d0/go.mod h1:x6AKhvSSexNrVSrViXSHUEbICjmGXhtgABaHIySUSGw=github.com/whyrusleeping/cbor-gen v0.2.1-0.20241030202151-b7a6831be65e h1:28X54ciEwwUxyHn9yrZfl5ojgF4CBNLWX7LR0rvBkf4=github.com/whyrusleeping/cbor-gen v0.2.1-0.20241030202151-b7a6831be65e/go.mod h1:pM99HXyEbSQHcosHc0iW7YFmwnscr+t9Te4ibko05so=+github.com/xeipuuv/gojsonpointer v0.0.0-20180127040702-4e3ac2762d5f h1:J9EGpcZtP0E/raorCMxlFGSTBrsSlaDGf3jU/qvAE2c=+github.com/xeipuuv/gojsonpointer v0.0.0-20180127040702-4e3ac2762d5f/go.mod h1:N2zxlSyiKSe5eX1tZViRH5QA0qijqEDrYZiPEAiq3wU=+github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415 h1:EzJWgHovont7NscjpAxXsDA8S8BMYve8Y5+7cuRE7R0=+github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415/go.mod h1:GwrjFmJcFw6At/Gs6z4yjiIwzuJ1/+UwLxMQDVQXShQ=
+54
internal/api/handlers/aggregator/errors.go
+54
internal/api/handlers/aggregator/errors.go
···
···+writeError(w, http.StatusNotImplemented, "NotImplemented", "This feature is not yet available (Phase 2)")
+194
internal/api/handlers/aggregator/get_services.go
+194
internal/api/handlers/aggregator/get_services.go
···
···+// GET /xrpc/social.coves.aggregator.getServices?dids=did:plc:abc123,did:plc:def456&detailed=true+// AggregatorViewDetailed matches social.coves.aggregator.defs#aggregatorViewDetailed (with stats)+// toAggregatorViewDetailed converts domain model to detailed aggregatorViewDetailed (with stats)
+173
internal/api/handlers/aggregator/list_for_community.go
+173
internal/api/handlers/aggregator/list_for_community.go
···
···+// GET /xrpc/social.coves.aggregator.listForCommunity?community=did:plc:xyz789&enabledOnly=true&limit=50&cursor=xyz+func (h *ListForCommunityHandler) HandleListForCommunity(w http.ResponseWriter, r *http.Request) {+// Note: Community handle/name fields will be empty until we integrate with communities service+response.Aggregators = append(response.Aggregators, toAuthorizationView(auth, req.CommunityDID))+func (h *ListForCommunityHandler) parseRequest(r *http.Request) (aggregators.ListForCommunityRequest, string, error) {+CommunityHandle *string `json:"communityHandle,omitempty"` // Optional: populated when communities service integration is complete+CommunityName *string `json:"communityName,omitempty"` // Optional: populated when communities service integration is complete+// communityHandle and communityName are left nil until communities service integration is complete+func toAuthorizationView(auth *aggregators.Authorization, communityDID string) AuthorizationView {
+6
internal/api/handlers/post/errors.go
+6
internal/api/handlers/post/errors.go
······
+8
internal/api/middleware/auth.go
+8
internal/api/middleware/auth.go
···
+39
internal/api/routes/aggregator.go
+39
internal/api/routes/aggregator.go
···
···+// GET /xrpc/social.coves.aggregator.getAuthorizations?aggregatorDid=did:plc:abc&enabledOnly=true+r.Get("/xrpc/social.coves.aggregator.getAuthorizations", getAuthorizationsHandler.HandleGetAuthorizations)+// GET /xrpc/social.coves.aggregator.listForCommunity?communityDid=did:plc:xyz&enabledOnly=true+r.Get("/xrpc/social.coves.aggregator.listForCommunity", listForCommunityHandler.HandleListForCommunity)
+347
internal/atproto/jetstream/aggregator_consumer.go
+347
internal/atproto/jetstream/aggregator_consumer.go
···
···+// Following Bluesky's pattern: feed generators (app.bsky.feed.generator) and labelers (app.bsky.labeler.service)+func (c *AggregatorEventConsumer) HandleEvent(ctx context.Context, event *JetstreamEvent) error {+// - social.coves.aggregator.service: Service declaration (in aggregator's own repo, rkey="self")+// Service declarations are stored at: at://aggregator_did/social.coves.aggregator.service/self+func (c *AggregatorEventConsumer) handleServiceDeclaration(ctx context.Context, did string, commit *CommitEvent) error {+// Authorizations are stored at: at://community_did/social.coves.aggregator.authorization/{rkey}+func (c *AggregatorEventConsumer) handleAuthorization(ctx context.Context, communityDID string, commit *CommitEvent) error {+func (c *AggregatorEventConsumer) upsertAggregator(ctx context.Context, did string, commit *CommitEvent) error {+log.Printf("[AGGREGATOR-CONSUMER] Aggregator not found for deletion: %s (already deleted?)", did)+func (c *AggregatorEventConsumer) upsertAuthorization(ctx context.Context, communityDID string, commit *CommitEvent) error {+uri := fmt.Sprintf("at://%s/social.coves.aggregator.authorization/%s", communityDID, commit.RKey)+log.Printf("[AGGREGATOR-CONSUMER] Indexed authorization: community=%s, aggregator=%s, enabled=%v",+func (c *AggregatorEventConsumer) deleteAuthorization(ctx context.Context, communityDID string, commit *CommitEvent) error {+uri := fmt.Sprintf("at://%s/social.coves.aggregator.authorization/%s", communityDID, commit.RKey)+log.Printf("[AGGREGATOR-CONSUMER] Authorization not found for deletion: %s (already deleted?)", uri)+Avatar map[string]interface{} `json:"avatar,omitempty"` // Blob reference (CID will be extracted)+DisabledAt string `json:"disabledAt,omitempty"` // When authorization was disabled (for modlog/audit)+func parseAggregatorAuthorization(record interface{}) (*AggregatorAuthorizationRecord, error) {
+136
internal/atproto/jetstream/aggregator_jetstream_connector.go
+136
internal/atproto/jetstream/aggregator_jetstream_connector.go
···
···+// AggregatorJetstreamConnector handles WebSocket connection to Jetstream for aggregator events+// NewAggregatorJetstreamConnector creates a new Jetstream WebSocket connector for aggregator events+func NewAggregatorJetstreamConnector(consumer *AggregatorEventConsumer, wsURL string) *AggregatorJetstreamConnector {
+97
internal/core/aggregators/aggregator.go
+97
internal/core/aggregators/aggregator.go
···
···+ConfigSchema []byte `json:"configSchema,omitempty" db:"config_schema"` // JSON Schema for configuration (JSONB)+MaintainerDID string `json:"maintainerDid,omitempty" db:"maintainer_did"` // Contact for support/issues+SourceURL string `json:"sourceUrl,omitempty" db:"source_url"` // Source code URL (transparency)+CommunitiesUsing int `json:"communitiesUsing" db:"communities_using"` // Auto-updated by trigger+CreatedAt time.Time `json:"createdAt" db:"created_at"` // When aggregator was created (from lexicon)+RecordURI string `json:"recordUri,omitempty" db:"record_uri"` // at://did/social.coves.aggregator.service/self+// Stored in community's repository: at://community_did/social.coves.aggregator.authorization/{rkey}+DisabledBy string `json:"disabledBy,omitempty" db:"disabled_by"` // Moderator DID who disabled it+DisabledAt *time.Time `json:"disabledAt,omitempty" db:"disabled_at"` // When authorization was disabled (for modlog/audit)+RecordURI string `json:"recordUri,omitempty" db:"record_uri"` // at://community_did/social.coves.aggregator.authorization/{rkey}
+63
internal/core/aggregators/errors.go
+63
internal/core/aggregators/errors.go
···
···+ErrNotImplemented = errors.New("feature not yet implemented") // For Phase 2 write-forward operations+return errors.As(err, &validationErr) || errors.Is(err, ErrInvalidConfig) || errors.Is(err, ErrConfigSchemaValidation)
+62
internal/core/aggregators/interfaces.go
+62
internal/core/aggregators/interfaces.go
···
···+GetAggregatorsByDIDs(ctx context.Context, dids []string) ([]*Aggregator, error) // Bulk fetch to avoid N+1 queries+IsAggregator(ctx context.Context, did string) (bool, error) // Fast check for post creation handler+GetAuthorization(ctx context.Context, aggregatorDID, communityDID string) (*Authorization, error)+GetAuthorizationByURI(ctx context.Context, recordURI string) (*Authorization, error) // For Jetstream delete operations+DeleteAuthorizationByURI(ctx context.Context, recordURI string) error // For Jetstream delete operations+ListAuthorizationsForAggregator(ctx context.Context, aggregatorDID string, enabledOnly bool, limit, offset int) ([]*Authorization, error)+ListAuthorizationsForCommunity(ctx context.Context, communityDID string, enabledOnly bool, limit, offset int) ([]*Authorization, error)+IsAuthorized(ctx context.Context, aggregatorDID, communityDID string) (bool, error) // Fast check: enabled=true+RecordAggregatorPost(ctx context.Context, aggregatorDID, communityDID, postURI, postCID string) error+CountRecentPosts(ctx context.Context, aggregatorDID, communityDID string, since time.Time) (int, error)+GetRecentPosts(ctx context.Context, aggregatorDID, communityDID string, since time.Time) ([]*AggregatorPost, error)+GetAuthorizationsForAggregator(ctx context.Context, req GetAuthorizationsRequest) ([]*Authorization, error)+ListAggregatorsForCommunity(ctx context.Context, req ListForCommunityRequest) ([]*Authorization, error)+// Authorization management (write-forward: Service -> PDS -> Firehose -> Consumer -> Repository)+ValidateAggregatorPost(ctx context.Context, aggregatorDID, communityDID string) error // Checks authorization + rate limits+IsAggregator(ctx context.Context, did string) (bool, error) // Check if DID is a registered aggregator+RecordAggregatorPost(ctx context.Context, aggregatorDID, communityDID, postURI, postCID string) error
+360
internal/core/aggregators/service.go
+360
internal/core/aggregators/service.go
···
···+RateLimitMaxPosts = 10 // Conservative limit for alpha: 10 posts/hour per community (prevents spam while allowing real-time updates)+MaxQueryLimit = 100 // Prevent abuse while allowing batch operations (e.g., fetching multiple aggregators at once)+func (s *aggregatorService) GetAggregator(ctx context.Context, did string) (*Aggregator, error) {+func (s *aggregatorService) GetAggregators(ctx context.Context, dids []string) ([]*Aggregator, error) {+func (s *aggregatorService) ListAggregators(ctx context.Context, limit, offset int) ([]*Aggregator, error) {+func (s *aggregatorService) GetAuthorizationsForAggregator(ctx context.Context, req GetAuthorizationsRequest) ([]*Authorization, error) {+return s.repo.ListAuthorizationsForAggregator(ctx, req.AggregatorDID, req.EnabledOnly, req.Limit, req.Offset)+func (s *aggregatorService) ListAggregatorsForCommunity(ctx context.Context, req ListForCommunityRequest) ([]*Authorization, error) {+return s.repo.ListAuthorizationsForCommunity(ctx, req.CommunityDID, req.EnabledOnly, req.Limit, req.Offset)+func (s *aggregatorService) EnableAggregator(ctx context.Context, req EnableAggregatorRequest) (*Authorization, error) {+func (s *aggregatorService) DisableAggregator(ctx context.Context, req DisableAggregatorRequest) (*Authorization, error) {+func (s *aggregatorService) UpdateAggregatorConfig(ctx context.Context, req UpdateConfigRequest) (*Authorization, error) {+func (s *aggregatorService) ValidateAggregatorPost(ctx context.Context, aggregatorDID, communityDID string) error {+func (s *aggregatorService) RecordAggregatorPost(ctx context.Context, aggregatorDID, communityDID, postURI, postCID string) error {+return NewValidationError("post_tracking", "aggregatorDID, communityDID, postURI, and postCID are required")+func (s *aggregatorService) validateEnableRequest(ctx context.Context, req EnableAggregatorRequest) error {+// membership, err := s.communityService.GetMembership(ctx, req.EnabledByDID, req.CommunityDID)+func (s *aggregatorService) validateDisableRequest(ctx context.Context, req DisableAggregatorRequest) error {+func (s *aggregatorService) validateUpdateConfigRequest(ctx context.Context, req UpdateConfigRequest) error {+func (s *aggregatorService) validateConfig(config map[string]interface{}, schemaBytes []byte) error {
+3
internal/core/posts/errors.go
+3
internal/core/posts/errors.go
···
+88
-22
internal/core/posts/service.go
+88
-22
internal/core/posts/service.go
······func (s *postService) CreatePost(ctx context.Context, req CreatePostRequest) (*CreatePostResponse, error) {·········-log.Printf("[POST-CREATE] Author: %s, Community: %s, URI: %s", req.AuthorDID, communityDID, uri)
······+// aggregatorService can be nil if aggregator support is not needed (e.g., in tests or minimal setups)func (s *postService) CreatePost(ctx context.Context, req CreatePostRequest) (*CreatePostResponse, error) {+log.Printf("[SECURITY] DID mismatch: authenticated=%s, request=%s", authenticatedDID, req.AuthorDID)+// If aggregatorService is nil (tests or environments without aggregators), treat all posts as user posts······+log.Printf("[POST-CREATE] Aggregator detected: %s posting to community: %s", req.AuthorDID, communityDID)+if err := s.aggregatorService.ValidateAggregatorPost(ctx, req.AuthorDID, communityDID); err != nil {···+if err := s.aggregatorService.RecordAggregatorPost(ctx, req.AuthorDID, communityDID, uri, cid); err != nil {+log.Printf("[POST-CREATE] Warning: failed to record aggregator post for rate limiting: %v", err)
+214
internal/db/migrations/012_create_aggregators_tables.sql
+214
internal/db/migrations/012_create_aggregators_tables.sql
···
···+-- Source: Aggregator's own repository (at://aggregator_did/social.coves.aggregator.service/self)+created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), -- When the aggregator service was created (from lexicon createdAt field)+communities_using INTEGER NOT NULL DEFAULT 0, -- Count of communities with enabled authorizations+COMMENT ON TABLE aggregators IS 'Aggregator service declarations indexed from social.coves.aggregator.service records';+COMMENT ON COLUMN aggregators.config_schema IS 'JSON Schema defining what config options communities can set';+COMMENT ON COLUMN aggregators.created_at IS 'When the aggregator service was created (from lexicon record createdAt field)';+COMMENT ON COLUMN aggregators.communities_using IS 'Cached count of communities with enabled=true authorizations';+-- Source: Community's repository (at://community_did/social.coves.aggregator.authorization/rkey)+CONSTRAINT fk_aggregator FOREIGN KEY (aggregator_did) REFERENCES aggregators(did) ON DELETE CASCADE,+CONSTRAINT fk_community FOREIGN KEY (community_did) REFERENCES communities(did) ON DELETE CASCADE+CREATE INDEX idx_aggregator_auth_agg_enabled ON aggregator_authorizations(aggregator_did, enabled) WHERE enabled = true;+CREATE INDEX idx_aggregator_auth_comm_enabled ON aggregator_authorizations(community_did, enabled) WHERE enabled = true;+CREATE INDEX idx_aggregator_auth_lookup ON aggregator_authorizations(aggregator_did, community_did, enabled);+CREATE INDEX idx_aggregator_auth_agg_did ON aggregator_authorizations(aggregator_did, created_at DESC);+CREATE INDEX idx_aggregator_auth_comm_did ON aggregator_authorizations(community_did, created_at DESC);+COMMENT ON TABLE aggregator_authorizations IS 'Community authorizations for aggregators indexed from social.coves.aggregator.authorization records';+COMMENT ON COLUMN aggregator_authorizations.config IS 'Community-specific config, validated against aggregators.config_schema';+COMMENT ON INDEX idx_aggregator_auth_lookup IS 'CRITICAL: Fast lookup for post creation authorization checks';+CONSTRAINT fk_aggregator_posts_agg FOREIGN KEY (aggregator_did) REFERENCES aggregators(did) ON DELETE CASCADE,+CONSTRAINT fk_aggregator_posts_comm FOREIGN KEY (community_did) REFERENCES communities(did) ON DELETE CASCADE+CREATE INDEX idx_aggregator_posts_rate_limit ON aggregator_posts(aggregator_did, community_did, created_at DESC);+CREATE INDEX idx_aggregator_posts_agg_did ON aggregator_posts(aggregator_did, created_at DESC);+CREATE INDEX idx_aggregator_posts_comm_did ON aggregator_posts(community_did, created_at DESC);+COMMENT ON TABLE aggregator_posts IS 'AppView-only tracking of posts created by aggregators for rate limiting and stats';+COMMENT ON INDEX idx_aggregator_posts_rate_limit IS 'CRITICAL: Fast rate limit checks (posts in last hour per community)';+COMMENT ON FUNCTION update_aggregator_communities_count IS 'Maintains aggregators.communities_using count when authorizations change';+COMMENT ON FUNCTION update_aggregator_posts_count IS 'Maintains aggregators.posts_created count when posts are tracked';+DROP TRIGGER IF EXISTS trigger_update_aggregator_communities_count ON aggregator_authorizations;
+813
internal/db/postgres/aggregator_repo.go
+813
internal/db/postgres/aggregator_repo.go
···
···+func (r *postgresAggregatorRepo) CreateAggregator(ctx context.Context, agg *aggregators.Aggregator) error {+func (r *postgresAggregatorRepo) GetAggregator(ctx context.Context, did string) (*aggregators.Aggregator, error) {+func (r *postgresAggregatorRepo) GetAggregatorsByDIDs(ctx context.Context, dids []string) ([]*aggregators.Aggregator, error) {+func (r *postgresAggregatorRepo) UpdateAggregator(ctx context.Context, agg *aggregators.Aggregator) error {+func (r *postgresAggregatorRepo) ListAggregators(ctx context.Context, limit, offset int) ([]*aggregators.Aggregator, error) {+func (r *postgresAggregatorRepo) IsAggregator(ctx context.Context, did string) (bool, error) {+func (r *postgresAggregatorRepo) CreateAuthorization(ctx context.Context, auth *aggregators.Authorization) error {+func (r *postgresAggregatorRepo) GetAuthorization(ctx context.Context, aggregatorDID, communityDID string) (*aggregators.Authorization, error) {+// GetAuthorizationByURI retrieves an authorization by record URI (for Jetstream delete operations)+func (r *postgresAggregatorRepo) GetAuthorizationByURI(ctx context.Context, recordURI string) (*aggregators.Authorization, error) {+func (r *postgresAggregatorRepo) UpdateAuthorization(ctx context.Context, auth *aggregators.Authorization) error {+func (r *postgresAggregatorRepo) DeleteAuthorization(ctx context.Context, aggregatorDID, communityDID string) error {+query := `DELETE FROM aggregator_authorizations WHERE aggregator_did = $1 AND community_did = $2`+// DeleteAuthorizationByURI removes an authorization by record URI (for Jetstream delete operations)+func (r *postgresAggregatorRepo) DeleteAuthorizationByURI(ctx context.Context, recordURI string) error {+func (r *postgresAggregatorRepo) ListAuthorizationsForAggregator(ctx context.Context, aggregatorDID string, enabledOnly bool, limit, offset int) ([]*aggregators.Authorization, error) {+func (r *postgresAggregatorRepo) ListAuthorizationsForCommunity(ctx context.Context, communityDID string, enabledOnly bool, limit, offset int) ([]*aggregators.Authorization, error) {+func (r *postgresAggregatorRepo) IsAuthorized(ctx context.Context, aggregatorDID, communityDID string) (bool, error) {+func (r *postgresAggregatorRepo) RecordAggregatorPost(ctx context.Context, aggregatorDID, communityDID, postURI, postCID string) error {+func (r *postgresAggregatorRepo) CountRecentPosts(ctx context.Context, aggregatorDID, communityDID string, since time.Time) (int, error) {+func (r *postgresAggregatorRepo) GetRecentPosts(ctx context.Context, aggregatorDID, communityDID string, since time.Time) ([]*aggregators.AggregatorPost, error) {
+864
tests/integration/aggregator_e2e_test.go
+864
tests/integration/aggregator_e2e_test.go
···
···+// 1. Service Declaration: Create aggregator account → Write service record → Jetstream → AppView DB+// 2. Authorization: Create community account → Write authorization record → Jetstream → AppView DB+// 3. Post Creation: Aggregator creates post → Validates authorization + rate limits → PDS → Jetstream → AppView+// - Simulated Jetstream events (for test speed - testing AppView indexing, not Jetstream itself)+communityService := communities.NewCommunityService(communityRepo, "http://localhost:3001", "did:web:test.coves.social", "coves.social", nil)+postService := posts.NewPostService(postRepo, communityService, aggregatorService, "http://localhost:3001")+authMiddleware := middleware.NewAtProtoAuthMiddleware(nil, true) // Skip JWT verification for testing+var aggregatorDID, aggregatorToken, aggregatorHandle, communityDID, communityToken, authorizationRkey string+t.Run("1. Service Declaration - PDS Account → Write Record → Jetstream → AppView DB", func(t *testing.T) {+aggregatorToken, aggregatorDID, err = createPDSAccount(pdsURL, aggregatorHandle, email, password)+uri, cid, err := writePDSRecord(pdsURL, aggregatorToken, aggregatorDID, "social.coves.aggregator.service", "self", serviceRecord)+// STEP 4: Process through Jetstream consumer (simulates what happens when Jetstream broadcasts)+assert.Equal(t, fmt.Sprintf("at://%s/social.coves.aggregator.service/self", aggregatorDID), indexedAgg.RecordURI)+t.Run("2. Authorization - Community Account → PDS → Jetstream → AppView DB", func(t *testing.T) {+communityToken, communityDID, err = createPDSAccount(pdsURL, communityHandle, communityEmail, communityPassword)+authURI, authCID, err := writePDSRecord(pdsURL, communityToken, communityDID, "social.coves.aggregator.authorization", "", authRecord)+authorizationRkey = strings.Split(authURI, "/")[4] // Extract rkey from URI and store for later+t.Run("3. Post Creation - Aggregator → Validation → PDS → Jetstream → AppView", func(t *testing.T) {+req := httptest.NewRequest("POST", "/xrpc/social.coves.post.create", bytes.NewReader(reqJSON))+require.Equal(t, http.StatusOK, rr.Code, "Handler should return 200 OK, body: %s", rr.Body.String())+req := httptest.NewRequest("POST", "/xrpc/social.coves.post.create", bytes.NewReader(reqJSON))+req := httptest.NewRequest("POST", "/xrpc/social.coves.post.create", bytes.NewReader(reqJSON))+req := httptest.NewRequest("GET", fmt.Sprintf("/xrpc/social.coves.aggregator.getServices?dids=%s", aggregatorDID), nil)+req := httptest.NewRequest("GET", fmt.Sprintf("/xrpc/social.coves.aggregator.getServices?dids=%s&detailed=true", aggregatorDID), nil)+req := httptest.NewRequest("GET", fmt.Sprintf("/xrpc/social.coves.aggregator.getAuthorizations?aggregatorDid=%s", aggregatorDID), nil)+req := httptest.NewRequest("GET", fmt.Sprintf("/xrpc/social.coves.aggregator.listForCommunity?community=%s", communityDID), nil)+req := httptest.NewRequest("POST", "/xrpc/social.coves.post.create", bytes.NewReader(reqJSON))+// Error message format from aggregators.ErrNotAuthorized: "aggregator not authorized for this community"+req := httptest.NewRequest("POST", "/xrpc/social.coves.post.create", bytes.NewReader(reqJSON))
+955
tests/integration/aggregator_test.go
+955
tests/integration/aggregator_test.go
···
···+RecordURI: fmt.Sprintf("at://%s/social.coves.aggregator.authorization/enabled", communityDID),+RecordURI: fmt.Sprintf("at://%s/social.coves.aggregator.authorization/disabled", communityDID2),+// TestAggregatorService_PostCreationIntegration tests the full post creation flow with aggregators+aggService := aggregators.NewAggregatorService(aggRepo, nil) // nil community service for this test+if err := aggRepo.RecordAggregatorPost(ctx, aggregatorDID, communityDID, postURI, "bafy123"); err != nil {+if err := aggRepo.RecordAggregatorPost(ctx, aggregatorDID, communityDID, postURI, "bafy123"); err != nil {+RecordURI: fmt.Sprintf("at://%s/social.coves.aggregator.authorization/auth%d", communityDID, i),+if err := aggRepo.RecordAggregatorPost(ctx, aggregatorDID, communityDID, postURI, "bafy123"); err != nil {+// TestAggregatorAuthorization_DisabledAtField tests that disabledAt is properly stored and retrieved+t.Errorf("Expected disabledAt to be nil for enabled authorization, got %v", *retrieved.DisabledAt)
+1
-1
tests/integration/community_identifier_resolution_test.go
+1
-1
tests/integration/community_identifier_resolution_test.go
+143
tests/integration/helpers.go
+143
tests/integration/helpers.go
······
······+func createPDSAccount(pdsURL, handle, email, password string) (accessToken, did string, err error) {+return "", "", fmt.Errorf("account creation failed (status %d, failed to read body: %w)", resp.StatusCode, readErr)+return "", "", fmt.Errorf("account creation failed (status %d): %s", resp.StatusCode, string(body))+func writePDSRecord(pdsURL, accessToken, repo, collection, rkey string, record interface{}) (uri, cid string, err error) {+req, reqErr := http.NewRequest("POST", pdsURL+"/xrpc/com.atproto.repo.createRecord", bytes.NewBuffer(reqJSON))+return "", "", fmt.Errorf("record creation failed (status %d, failed to read body: %w)", resp.StatusCode, readErr)+return "", "", fmt.Errorf("record creation failed (status %d): %s", resp.StatusCode, string(body))
+1
-1
tests/integration/post_creation_test.go
+1
-1
tests/integration/post_creation_test.go
+1
-44
tests/integration/post_e2e_test.go
+1
-44
tests/integration/post_e2e_test.go
···············
············+postService := posts.NewPostService(postRepo, communityService, nil, pdsURL) // nil aggregatorService for user-only tests···
+88
-2
tests/integration/post_handler_test.go
+88
-2
tests/integration/post_handler_test.go
·········
···+postService := posts.NewPostService(postRepo, communityService, nil, "http://localhost:3001") // nil aggregatorService for user-only tests···+postService := posts.NewPostService(postRepo, communityService, nil, "http://localhost:3001") // nil aggregatorService for user-only tests···+// TestPostService_DIDValidationSecurity tests service-layer DID validation (defense-in-depth)