diff --git a/go.mod b/go.mod index f6f96bc..887d290 100644 --- a/go.mod +++ b/go.mod @@ -72,6 +72,9 @@ require ( github.com/sethvargo/go-retry v0.3.0 // indirect github.com/spaolacci/murmur3 v1.1.0 // indirect github.com/whyrusleeping/cbor-gen v0.2.1-0.20241030202151-b7a6831be65e // indirect + github.com/xeipuuv/gojsonpointer v0.0.0-20180127040702-4e3ac2762d5f // indirect + github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415 // indirect + github.com/xeipuuv/gojsonschema v1.2.0 // indirect gitlab.com/yawning/secp256k1-voi v0.0.0-20230925100816-f2616030848b // indirect gitlab.com/yawning/tuplehash v0.0.0-20230713102510-df83abbf9a02 // indirect go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.46.1 // indirect diff --git a/go.sum b/go.sum index 2252fdf..6707f9e 100644 --- a/go.sum +++ b/go.sum @@ -187,6 +187,12 @@ github.com/warpfork/go-wish v0.0.0-20220906213052-39a1cc7a02d0 h1:GDDkbFiaK8jsSD 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= +github.com/xeipuuv/gojsonschema v1.2.0 h1:LhYJRs+L4fBtjZUfuSZIKGeVu0QRy8e5Xi7D17UxZ74= +github.com/xeipuuv/gojsonschema v1.2.0/go.mod h1:anYRn/JVcOK2ZgGU+IjEV4nwlhoK5sQluxsYJ78Id3Y= github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k= diff --git a/internal/core/aggregators/aggregator.go b/internal/core/aggregators/aggregator.go new file mode 100644 index 0000000..0f4e8d4 --- /dev/null +++ b/internal/core/aggregators/aggregator.go @@ -0,0 +1,97 @@ +package aggregators + +import "time" + +// Aggregator represents a service declaration record indexed from the firehose +// Aggregators are autonomous services that can post content to communities after authorization +// Following Bluesky's pattern: app.bsky.feed.generator and app.bsky.labeler.service +type Aggregator struct { + DID string `json:"did" db:"did"` // Aggregator's DID (primary key) + DisplayName string `json:"displayName" db:"display_name"` // Human-readable name + Description string `json:"description,omitempty" db:"description"` // What the aggregator does + AvatarURL string `json:"avatarUrl,omitempty" db:"avatar_url"` // Optional avatar image URL + 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 + PostsCreated int `json:"postsCreated" db:"posts_created"` // Auto-updated by trigger + CreatedAt time.Time `json:"createdAt" db:"created_at"` // When aggregator was created (from lexicon) + IndexedAt time.Time `json:"indexedAt" db:"indexed_at"` // When we indexed this record + RecordURI string `json:"recordUri,omitempty" db:"record_uri"` // at://did/social.coves.aggregator.service/self + RecordCID string `json:"recordCid,omitempty" db:"record_cid"` // Content hash +} + +// Authorization represents a community's authorization for an aggregator +// Stored in community's repository: at://community_did/social.coves.aggregator.authorization/{rkey} +type Authorization struct { + ID int `json:"id" db:"id"` // Database ID + AggregatorDID string `json:"aggregatorDid" db:"aggregator_did"` // Which aggregator + CommunityDID string `json:"communityDid" db:"community_did"` // Which community + Enabled bool `json:"enabled" db:"enabled"` // Current status + Config []byte `json:"config,omitempty" db:"config"` // Aggregator-specific config (JSONB) + CreatedBy string `json:"createdBy,omitempty" db:"created_by"` // Moderator DID who enabled it + DisabledBy string `json:"disabledBy,omitempty" db:"disabled_by"` // Moderator DID who disabled it + CreatedAt time.Time `json:"createdAt" db:"created_at"` // When authorization was created + DisabledAt *time.Time `json:"disabledAt,omitempty" db:"disabled_at"` // When authorization was disabled (for modlog/audit) + IndexedAt time.Time `json:"indexedAt" db:"indexed_at"` // When we indexed this record + RecordURI string `json:"recordUri,omitempty" db:"record_uri"` // at://community_did/social.coves.aggregator.authorization/{rkey} + RecordCID string `json:"recordCid,omitempty" db:"record_cid"` // Content hash +} + +// AggregatorPost represents tracking of posts created by aggregators +// AppView-only table for rate limiting and statistics +type AggregatorPost struct { + ID int `json:"id" db:"id"` + AggregatorDID string `json:"aggregatorDid" db:"aggregator_did"` + CommunityDID string `json:"communityDid" db:"community_did"` + PostURI string `json:"postUri" db:"post_uri"` + PostCID string `json:"postCid" db:"post_cid"` + CreatedAt time.Time `json:"createdAt" db:"created_at"` +} + +// EnableAggregatorRequest represents input for enabling an aggregator in a community +type EnableAggregatorRequest struct { + CommunityDID string `json:"communityDid"` // Which community (resolved from identifier) + AggregatorDID string `json:"aggregatorDid"` // Which aggregator + Config map[string]interface{} `json:"config,omitempty"` // Aggregator-specific configuration + EnabledByDID string `json:"enabledByDid"` // Moderator making the change (from JWT) + EnabledByToken string `json:"-"` // User's access token for PDS write +} + +// DisableAggregatorRequest represents input for disabling an aggregator +type DisableAggregatorRequest struct { + CommunityDID string `json:"communityDid"` // Which community (resolved from identifier) + AggregatorDID string `json:"aggregatorDid"` // Which aggregator + DisabledByDID string `json:"disabledByDid"` // Moderator making the change (from JWT) + DisabledByToken string `json:"-"` // User's access token for PDS write +} + +// UpdateConfigRequest represents input for updating an aggregator's configuration +type UpdateConfigRequest struct { + CommunityDID string `json:"communityDid"` // Which community (resolved from identifier) + AggregatorDID string `json:"aggregatorDid"` // Which aggregator + Config map[string]interface{} `json:"config"` // New configuration + UpdatedByDID string `json:"updatedByDid"` // Moderator making the change (from JWT) + UpdatedByToken string `json:"-"` // User's access token for PDS write +} + +// GetServicesRequest represents query parameters for fetching aggregator details +type GetServicesRequest struct { + DIDs []string `json:"dids"` // List of aggregator DIDs to fetch +} + +// GetAuthorizationsRequest represents query parameters for listing authorizations +type GetAuthorizationsRequest struct { + AggregatorDID string `json:"aggregatorDid"` // Which aggregator + EnabledOnly bool `json:"enabledOnly,omitempty"` // Only return enabled authorizations + Limit int `json:"limit"` + Offset int `json:"offset"` +} + +// ListForCommunityRequest represents query parameters for listing aggregators for a community +type ListForCommunityRequest struct { + CommunityDID string `json:"communityDid"` // Which community (resolved from identifier) + EnabledOnly bool `json:"enabledOnly,omitempty"` // Only return enabled aggregators + Limit int `json:"limit"` + Offset int `json:"offset"` +} diff --git a/internal/core/aggregators/errors.go b/internal/core/aggregators/errors.go new file mode 100644 index 0000000..b2cbfa4 --- /dev/null +++ b/internal/core/aggregators/errors.go @@ -0,0 +1,63 @@ +package aggregators + +import ( + "errors" + "fmt" +) + +// Domain errors +var ( + ErrAggregatorNotFound = errors.New("aggregator not found") + ErrAuthorizationNotFound = errors.New("authorization not found") + ErrNotAuthorized = errors.New("aggregator not authorized for this community") + ErrAlreadyAuthorized = errors.New("aggregator already authorized for this community") + ErrRateLimitExceeded = errors.New("aggregator rate limit exceeded") + ErrInvalidConfig = errors.New("invalid aggregator configuration") + ErrConfigSchemaValidation = errors.New("configuration does not match aggregator's schema") + ErrNotModerator = errors.New("user is not a moderator of this community") + ErrNotImplemented = errors.New("feature not yet implemented") // For Phase 2 write-forward operations +) + +// ValidationError represents a validation error with field details +type ValidationError struct { + Field string + Message string +} + +func (e *ValidationError) Error() string { + return fmt.Sprintf("validation error: %s - %s", e.Field, e.Message) +} + +// NewValidationError creates a new validation error +func NewValidationError(field, message string) error { + return &ValidationError{ + Field: field, + Message: message, + } +} + +// Error classification helpers for handlers to map to HTTP status codes +func IsNotFound(err error) bool { + return errors.Is(err, ErrAggregatorNotFound) || errors.Is(err, ErrAuthorizationNotFound) +} + +func IsValidationError(err error) bool { + var validationErr *ValidationError + return errors.As(err, &validationErr) || errors.Is(err, ErrInvalidConfig) || errors.Is(err, ErrConfigSchemaValidation) +} + +func IsUnauthorized(err error) bool { + return errors.Is(err, ErrNotAuthorized) || errors.Is(err, ErrNotModerator) +} + +func IsConflict(err error) bool { + return errors.Is(err, ErrAlreadyAuthorized) +} + +func IsRateLimited(err error) bool { + return errors.Is(err, ErrRateLimitExceeded) +} + +func IsNotImplemented(err error) bool { + return errors.Is(err, ErrNotImplemented) +} diff --git a/internal/core/aggregators/interfaces.go b/internal/core/aggregators/interfaces.go new file mode 100644 index 0000000..1825acc --- /dev/null +++ b/internal/core/aggregators/interfaces.go @@ -0,0 +1,62 @@ +package aggregators + +import ( + "context" + "time" +) + +// Repository defines the interface for aggregator data persistence +// This is the AppView's indexed view of aggregators and authorizations from the firehose +type Repository interface { + // Aggregator CRUD (indexed from firehose) + CreateAggregator(ctx context.Context, aggregator *Aggregator) error + GetAggregator(ctx context.Context, did string) (*Aggregator, error) + GetAggregatorsByDIDs(ctx context.Context, dids []string) ([]*Aggregator, error) // Bulk fetch to avoid N+1 queries + UpdateAggregator(ctx context.Context, aggregator *Aggregator) error + DeleteAggregator(ctx context.Context, did string) error + ListAggregators(ctx context.Context, limit, offset int) ([]*Aggregator, error) + IsAggregator(ctx context.Context, did string) (bool, error) // Fast check for post creation handler + + // Authorization CRUD (indexed from firehose) + CreateAuthorization(ctx context.Context, auth *Authorization) error + GetAuthorization(ctx context.Context, aggregatorDID, communityDID string) (*Authorization, error) + GetAuthorizationByURI(ctx context.Context, recordURI string) (*Authorization, error) // For Jetstream delete operations + UpdateAuthorization(ctx context.Context, auth *Authorization) error + DeleteAuthorization(ctx context.Context, aggregatorDID, communityDID string) error + DeleteAuthorizationByURI(ctx context.Context, recordURI string) error // For Jetstream delete operations + + // Authorization queries + 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 + + // Post tracking (for rate limiting and stats) + 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) +} + +// Service defines the interface for aggregator business logic +// Coordinates between Repository, communities service, and PDS for write-forward +type Service interface { + // Aggregator queries (read from AppView) + GetAggregator(ctx context.Context, did string) (*Aggregator, error) + GetAggregators(ctx context.Context, dids []string) ([]*Aggregator, error) + ListAggregators(ctx context.Context, limit, offset int) ([]*Aggregator, error) + + // Authorization queries (read from AppView) + 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) + EnableAggregator(ctx context.Context, req EnableAggregatorRequest) (*Authorization, error) + DisableAggregator(ctx context.Context, req DisableAggregatorRequest) (*Authorization, error) + UpdateAggregatorConfig(ctx context.Context, req UpdateConfigRequest) (*Authorization, error) + + // Validation and authorization checks (used by post creation handler) + 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 + + // Post tracking (called after successful post creation) + RecordAggregatorPost(ctx context.Context, aggregatorDID, communityDID, postURI, postCID string) error +} diff --git a/internal/core/aggregators/service.go b/internal/core/aggregators/service.go new file mode 100644 index 0000000..5864eec --- /dev/null +++ b/internal/core/aggregators/service.go @@ -0,0 +1,360 @@ +package aggregators + +import ( + "Coves/internal/core/communities" + "context" + "encoding/json" + "fmt" + "time" + + "github.com/xeipuuv/gojsonschema" +) + +// Rate limit constants +const ( + RateLimitWindow = 1 * time.Hour // Rolling 1-hour window for rate limit enforcement + RateLimitMaxPosts = 10 // Conservative limit for alpha: 10 posts/hour per community (prevents spam while allowing real-time updates) + DefaultQueryLimit = 50 // Balance between UX (reasonable page size) and server load + MaxQueryLimit = 100 // Prevent abuse while allowing batch operations (e.g., fetching multiple aggregators at once) +) + +type aggregatorService struct { + repo Repository + communityService communities.Service +} + +// NewAggregatorService creates a new aggregator service +func NewAggregatorService(repo Repository, communityService communities.Service) Service { + return &aggregatorService{ + repo: repo, + communityService: communityService, + } +} + +// ===== Query Operations (Read from AppView) ===== + +// GetAggregator retrieves a single aggregator by DID +func (s *aggregatorService) GetAggregator(ctx context.Context, did string) (*Aggregator, error) { + if did == "" { + return nil, NewValidationError("did", "DID is required") + } + + return s.repo.GetAggregator(ctx, did) +} + +// GetAggregators retrieves multiple aggregators by DIDs +func (s *aggregatorService) GetAggregators(ctx context.Context, dids []string) ([]*Aggregator, error) { + if len(dids) == 0 { + return []*Aggregator{}, nil + } + + if len(dids) > MaxQueryLimit { + return nil, NewValidationError("dids", fmt.Sprintf("maximum %d DIDs allowed", MaxQueryLimit)) + } + + // Use bulk fetch to avoid N+1 queries + return s.repo.GetAggregatorsByDIDs(ctx, dids) +} + +// ListAggregators retrieves all aggregators with pagination +func (s *aggregatorService) ListAggregators(ctx context.Context, limit, offset int) ([]*Aggregator, error) { + // Apply defaults and limits + if limit <= 0 { + limit = DefaultQueryLimit + } + if limit > MaxQueryLimit { + limit = MaxQueryLimit + } + if offset < 0 { + offset = 0 + } + + return s.repo.ListAggregators(ctx, limit, offset) +} + +// GetAuthorizationsForAggregator retrieves all communities that authorized an aggregator +func (s *aggregatorService) GetAuthorizationsForAggregator(ctx context.Context, req GetAuthorizationsRequest) ([]*Authorization, error) { + if req.AggregatorDID == "" { + return nil, NewValidationError("aggregatorDid", "aggregator DID is required") + } + + // Apply defaults and limits + if req.Limit <= 0 { + req.Limit = DefaultQueryLimit + } + if req.Limit > MaxQueryLimit { + req.Limit = MaxQueryLimit + } + if req.Offset < 0 { + req.Offset = 0 + } + + return s.repo.ListAuthorizationsForAggregator(ctx, req.AggregatorDID, req.EnabledOnly, req.Limit, req.Offset) +} + +// ListAggregatorsForCommunity retrieves all aggregators authorized by a community +func (s *aggregatorService) ListAggregatorsForCommunity(ctx context.Context, req ListForCommunityRequest) ([]*Authorization, error) { + if req.CommunityDID == "" { + return nil, NewValidationError("communityDid", "community DID is required") + } + + // Apply defaults and limits + if req.Limit <= 0 { + req.Limit = DefaultQueryLimit + } + if req.Limit > MaxQueryLimit { + req.Limit = MaxQueryLimit + } + if req.Offset < 0 { + req.Offset = 0 + } + + return s.repo.ListAuthorizationsForCommunity(ctx, req.CommunityDID, req.EnabledOnly, req.Limit, req.Offset) +} + +// ===== Authorization Management (Write-forward to PDS) ===== + +// EnableAggregator creates an authorization record for an aggregator in a community +// Following Bluesky's pattern: similar to enabling a labeler or feed generator +// Note: This is a PLACEHOLDER for the write-forward implementation +// TODO: Implement actual XRPC write to community's PDS repository +func (s *aggregatorService) EnableAggregator(ctx context.Context, req EnableAggregatorRequest) (*Authorization, error) { + // Validate request + if err := s.validateEnableRequest(ctx, req); err != nil { + return nil, err + } + + // Verify aggregator exists + aggregator, err := s.repo.GetAggregator(ctx, req.AggregatorDID) + if err != nil { + return nil, err + } + + // Validate config against aggregator's schema if provided + if len(req.Config) > 0 && len(aggregator.ConfigSchema) > 0 { + if err := s.validateConfig(req.Config, aggregator.ConfigSchema); err != nil { + return nil, err + } + } + + // Check if already authorized + existing, err := s.repo.GetAuthorization(ctx, req.AggregatorDID, req.CommunityDID) + if err == nil && existing.Enabled { + return nil, ErrAlreadyAuthorized + } + + // TODO Phase 2: Write-forward to PDS + // For now, return placeholder response + // The actual implementation will: + // 1. Create authorization record in community's repository on PDS + // 2. Wait for Jetstream to index it + // 3. Return the indexed authorization + // + // Record structure: + // at://community_did/social.coves.aggregator.authorization/{rkey} + // { + // "$type": "social.coves.aggregator.authorization", + // "aggregator": req.AggregatorDID, + // "enabled": true, + // "config": req.Config, + // "createdBy": req.EnabledByDID, + // "createdAt": "2025-10-20T12:00:00Z" + // } + + return nil, ErrNotImplemented +} + +// DisableAggregator updates an authorization to disabled +// Note: This is a PLACEHOLDER for the write-forward implementation +func (s *aggregatorService) DisableAggregator(ctx context.Context, req DisableAggregatorRequest) (*Authorization, error) { + // Validate request + if err := s.validateDisableRequest(ctx, req); err != nil { + return nil, err + } + + // Verify authorization exists + auth, err := s.repo.GetAuthorization(ctx, req.AggregatorDID, req.CommunityDID) + if err != nil { + return nil, err + } + + if !auth.Enabled { + // Already disabled + return auth, nil + } + + // TODO Phase 2: Write-forward to PDS + // Update the authorization record with enabled=false + return nil, ErrNotImplemented +} + +// UpdateAggregatorConfig updates an aggregator's configuration +// Note: This is a PLACEHOLDER for the write-forward implementation +func (s *aggregatorService) UpdateAggregatorConfig(ctx context.Context, req UpdateConfigRequest) (*Authorization, error) { + // Validate request + if err := s.validateUpdateConfigRequest(ctx, req); err != nil { + return nil, err + } + + // Verify authorization exists + auth, err := s.repo.GetAuthorization(ctx, req.AggregatorDID, req.CommunityDID) + if err != nil { + return nil, err + } + + // Get aggregator for schema validation + aggregator, err := s.repo.GetAggregator(ctx, req.AggregatorDID) + if err != nil { + return nil, err + } + + // Validate new config against schema + if len(req.Config) > 0 && len(aggregator.ConfigSchema) > 0 { + if err := s.validateConfig(req.Config, aggregator.ConfigSchema); err != nil { + return nil, err + } + } + + // TODO Phase 2: Write-forward to PDS + // Update the authorization record with new config + return auth, ErrNotImplemented +} + +// ===== Validation and Authorization Checks ===== + +// ValidateAggregatorPost validates that an aggregator can post to a community +// Checks: 1) Authorization exists and is enabled, 2) Rate limit not exceeded +// This is called by the post creation handler BEFORE writing to PDS +func (s *aggregatorService) ValidateAggregatorPost(ctx context.Context, aggregatorDID, communityDID string) error { + // Check authorization exists and is enabled + authorized, err := s.repo.IsAuthorized(ctx, aggregatorDID, communityDID) + if err != nil { + return fmt.Errorf("failed to check authorization: %w", err) + } + if !authorized { + return ErrNotAuthorized + } + + // Check rate limit (10 posts per hour per community) + since := time.Now().Add(-RateLimitWindow) + recentPostCount, err := s.repo.CountRecentPosts(ctx, aggregatorDID, communityDID, since) + if err != nil { + return fmt.Errorf("failed to check rate limit: %w", err) + } + + if recentPostCount >= RateLimitMaxPosts { + return ErrRateLimitExceeded + } + + return nil +} + +// IsAggregator checks if a DID is a registered aggregator +// Fast check used by post creation handler +func (s *aggregatorService) IsAggregator(ctx context.Context, did string) (bool, error) { + if did == "" { + return false, nil + } + return s.repo.IsAggregator(ctx, did) +} + +// RecordAggregatorPost tracks a post created by an aggregator +// Called AFTER successful post creation to update statistics and rate limiting +func (s *aggregatorService) RecordAggregatorPost(ctx context.Context, aggregatorDID, communityDID, postURI, postCID string) error { + if aggregatorDID == "" || communityDID == "" || postURI == "" || postCID == "" { + return NewValidationError("post_tracking", "aggregatorDID, communityDID, postURI, and postCID are required") + } + + return s.repo.RecordAggregatorPost(ctx, aggregatorDID, communityDID, postURI, postCID) +} + +// ===== Validation Helpers ===== + +func (s *aggregatorService) validateEnableRequest(ctx context.Context, req EnableAggregatorRequest) error { + if req.AggregatorDID == "" { + return NewValidationError("aggregatorDid", "aggregator DID is required") + } + if req.CommunityDID == "" { + return NewValidationError("communityDid", "community DID is required") + } + if req.EnabledByDID == "" { + return NewValidationError("enabledByDid", "enabledByDID is required") + } + + // Verify user is a moderator of the community + // TODO: Implement moderator check + // membership, err := s.communityService.GetMembership(ctx, req.EnabledByDID, req.CommunityDID) + // if err != nil || !membership.IsModerator { + // return ErrNotModerator + // } + + return nil +} + +func (s *aggregatorService) validateDisableRequest(ctx context.Context, req DisableAggregatorRequest) error { + if req.AggregatorDID == "" { + return NewValidationError("aggregatorDid", "aggregator DID is required") + } + if req.CommunityDID == "" { + return NewValidationError("communityDid", "community DID is required") + } + if req.DisabledByDID == "" { + return NewValidationError("disabledByDid", "disabledByDID is required") + } + + // Verify user is a moderator of the community + // TODO: Implement moderator check + + return nil +} + +func (s *aggregatorService) validateUpdateConfigRequest(ctx context.Context, req UpdateConfigRequest) error { + if req.AggregatorDID == "" { + return NewValidationError("aggregatorDid", "aggregator DID is required") + } + if req.CommunityDID == "" { + return NewValidationError("communityDid", "community DID is required") + } + if req.UpdatedByDID == "" { + return NewValidationError("updatedByDid", "updatedByDID is required") + } + if len(req.Config) == 0 { + return NewValidationError("config", "config is required") + } + + // Verify user is a moderator of the community + // TODO: Implement moderator check + + return nil +} + +// validateConfig validates a config object against a JSON Schema +// Following Bluesky's pattern for feed generator configuration +func (s *aggregatorService) validateConfig(config map[string]interface{}, schemaBytes []byte) error { + // Parse schema + schemaLoader := gojsonschema.NewBytesLoader(schemaBytes) + + // Convert config to JSON bytes + configBytes, err := json.Marshal(config) + if err != nil { + return fmt.Errorf("failed to marshal config: %w", err) + } + configLoader := gojsonschema.NewBytesLoader(configBytes) + + // Validate + result, err := gojsonschema.Validate(schemaLoader, configLoader) + if err != nil { + return fmt.Errorf("failed to validate config: %w", err) + } + + if !result.Valid() { + // Collect validation errors + var errorMessages []string + for _, desc := range result.Errors() { + errorMessages = append(errorMessages, desc.String()) + } + return fmt.Errorf("%w: %s", ErrConfigSchemaValidation, errorMessages) + } + + return nil +} diff --git a/internal/db/postgres/aggregator_repo.go b/internal/db/postgres/aggregator_repo.go new file mode 100644 index 0000000..28e4582 --- /dev/null +++ b/internal/db/postgres/aggregator_repo.go @@ -0,0 +1,813 @@ +package postgres + +import ( + "Coves/internal/core/aggregators" + "context" + "database/sql" + "fmt" + "strings" + "time" +) + +type postgresAggregatorRepo struct { + db *sql.DB +} + +// NewAggregatorRepository creates a new PostgreSQL aggregator repository +func NewAggregatorRepository(db *sql.DB) aggregators.Repository { + return &postgresAggregatorRepo{db: db} +} + +// ===== Aggregator CRUD Operations ===== + +// CreateAggregator indexes a new aggregator service declaration from the firehose +func (r *postgresAggregatorRepo) CreateAggregator(ctx context.Context, agg *aggregators.Aggregator) error { + query := ` + INSERT INTO aggregators ( + did, display_name, description, avatar_url, config_schema, + maintainer_did, source_url, created_at, indexed_at, record_uri, record_cid + ) VALUES ( + $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11 + ) + ON CONFLICT (did) DO UPDATE SET + display_name = EXCLUDED.display_name, + description = EXCLUDED.description, + avatar_url = EXCLUDED.avatar_url, + config_schema = EXCLUDED.config_schema, + maintainer_did = EXCLUDED.maintainer_did, + source_url = EXCLUDED.source_url, + created_at = EXCLUDED.created_at, + indexed_at = EXCLUDED.indexed_at, + record_uri = EXCLUDED.record_uri, + record_cid = EXCLUDED.record_cid` + + var configSchema interface{} + if len(agg.ConfigSchema) > 0 { + configSchema = agg.ConfigSchema + } else { + configSchema = nil + } + + _, err := r.db.ExecContext(ctx, query, + agg.DID, + agg.DisplayName, + nullString(agg.Description), + nullString(agg.AvatarURL), + configSchema, + nullString(agg.MaintainerDID), + nullString(agg.SourceURL), + agg.CreatedAt, + agg.IndexedAt, + nullString(agg.RecordURI), + nullString(agg.RecordCID), + ) + + if err != nil { + return fmt.Errorf("failed to create aggregator: %w", err) + } + + return nil +} + +// GetAggregator retrieves an aggregator by DID +func (r *postgresAggregatorRepo) GetAggregator(ctx context.Context, did string) (*aggregators.Aggregator, error) { + query := ` + SELECT + did, display_name, description, avatar_url, config_schema, + maintainer_did, source_url, communities_using, posts_created, + created_at, indexed_at, record_uri, record_cid + FROM aggregators + WHERE did = $1` + + agg := &aggregators.Aggregator{} + var description, avatarCID, maintainerDID, homepageURL, recordURI, recordCID sql.NullString + var configSchema []byte + + err := r.db.QueryRowContext(ctx, query, did).Scan( + &agg.DID, + &agg.DisplayName, + &description, + &avatarCID, + &configSchema, + &maintainerDID, + &homepageURL, + &agg.CommunitiesUsing, + &agg.PostsCreated, + &agg.CreatedAt, + &agg.IndexedAt, + &recordURI, + &recordCID, + ) + + if err == sql.ErrNoRows { + return nil, aggregators.ErrAggregatorNotFound + } + if err != nil { + return nil, fmt.Errorf("failed to get aggregator: %w", err) + } + + // Map nullable fields + agg.Description = description.String + agg.AvatarURL = avatarCID.String + agg.MaintainerDID = maintainerDID.String + agg.SourceURL = homepageURL.String + agg.RecordURI = recordURI.String + agg.RecordCID = recordCID.String + if configSchema != nil { + agg.ConfigSchema = configSchema + } + + return agg, nil +} + +// GetAggregatorsByDIDs retrieves multiple aggregators by DIDs in a single query (avoids N+1) +func (r *postgresAggregatorRepo) GetAggregatorsByDIDs(ctx context.Context, dids []string) ([]*aggregators.Aggregator, error) { + if len(dids) == 0 { + return []*aggregators.Aggregator{}, nil + } + + // Build IN clause with placeholders + placeholders := make([]string, len(dids)) + args := make([]interface{}, len(dids)) + for i, did := range dids { + placeholders[i] = fmt.Sprintf("$%d", i+1) + args[i] = did + } + + query := fmt.Sprintf(` + SELECT + did, display_name, description, avatar_url, config_schema, + maintainer_did, source_url, communities_using, posts_created, + created_at, indexed_at, record_uri, record_cid + FROM aggregators + WHERE did IN (%s)`, strings.Join(placeholders, ", ")) + + rows, err := r.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, fmt.Errorf("failed to get aggregators: %w", err) + } + defer rows.Close() + + var results []*aggregators.Aggregator + for rows.Next() { + agg := &aggregators.Aggregator{} + var description, avatarCID, maintainerDID, homepageURL, recordURI, recordCID sql.NullString + var configSchema []byte + + err := rows.Scan( + &agg.DID, + &agg.DisplayName, + &description, + &avatarCID, + &configSchema, + &maintainerDID, + &homepageURL, + &agg.CommunitiesUsing, + &agg.PostsCreated, + &agg.CreatedAt, + &agg.IndexedAt, + &recordURI, + &recordCID, + ) + if err != nil { + return nil, fmt.Errorf("failed to scan aggregator: %w", err) + } + + // Map nullable fields + agg.Description = description.String + agg.AvatarURL = avatarCID.String + agg.MaintainerDID = maintainerDID.String + agg.SourceURL = homepageURL.String + agg.RecordURI = recordURI.String + agg.RecordCID = recordCID.String + if configSchema != nil { + agg.ConfigSchema = configSchema + } + + results = append(results, agg) + } + + if err = rows.Err(); err != nil { + return nil, fmt.Errorf("error iterating aggregators: %w", err) + } + + return results, nil +} + +// UpdateAggregator updates an existing aggregator +func (r *postgresAggregatorRepo) UpdateAggregator(ctx context.Context, agg *aggregators.Aggregator) error { + query := ` + UPDATE aggregators SET + display_name = $2, + description = $3, + avatar_url = $4, + config_schema = $5, + maintainer_did = $6, + source_url = $7, + created_at = $8, + indexed_at = $9, + record_uri = $10, + record_cid = $11 + WHERE did = $1` + + var configSchema interface{} + if len(agg.ConfigSchema) > 0 { + configSchema = agg.ConfigSchema + } else { + configSchema = nil + } + + result, err := r.db.ExecContext(ctx, query, + agg.DID, + agg.DisplayName, + nullString(agg.Description), + nullString(agg.AvatarURL), + configSchema, + nullString(agg.MaintainerDID), + nullString(agg.SourceURL), + agg.CreatedAt, + agg.IndexedAt, + nullString(agg.RecordURI), + nullString(agg.RecordCID), + ) + + if err != nil { + return fmt.Errorf("failed to update aggregator: %w", err) + } + + rows, err := result.RowsAffected() + if err != nil { + return fmt.Errorf("failed to get rows affected: %w", err) + } + if rows == 0 { + return aggregators.ErrAggregatorNotFound + } + + return nil +} + +// DeleteAggregator removes an aggregator (cascade deletes authorizations and posts via FK) +func (r *postgresAggregatorRepo) DeleteAggregator(ctx context.Context, did string) error { + query := `DELETE FROM aggregators WHERE did = $1` + + result, err := r.db.ExecContext(ctx, query, did) + if err != nil { + return fmt.Errorf("failed to delete aggregator: %w", err) + } + + rows, err := result.RowsAffected() + if err != nil { + return fmt.Errorf("failed to get rows affected: %w", err) + } + if rows == 0 { + return aggregators.ErrAggregatorNotFound + } + + return nil +} + +// ListAggregators retrieves all aggregators with pagination +func (r *postgresAggregatorRepo) ListAggregators(ctx context.Context, limit, offset int) ([]*aggregators.Aggregator, error) { + query := ` + SELECT + did, display_name, description, avatar_url, config_schema, + maintainer_did, source_url, communities_using, posts_created, + created_at, indexed_at, record_uri, record_cid + FROM aggregators + ORDER BY communities_using DESC, display_name ASC + LIMIT $1 OFFSET $2` + + rows, err := r.db.QueryContext(ctx, query, limit, offset) + if err != nil { + return nil, fmt.Errorf("failed to list aggregators: %w", err) + } + defer rows.Close() + + var aggs []*aggregators.Aggregator + for rows.Next() { + agg := &aggregators.Aggregator{} + var description, avatarCID, maintainerDID, homepageURL, recordURI, recordCID sql.NullString + var configSchema []byte + + err := rows.Scan( + &agg.DID, + &agg.DisplayName, + &description, + &avatarCID, + &configSchema, + &maintainerDID, + &homepageURL, + &agg.CommunitiesUsing, + &agg.PostsCreated, + &agg.CreatedAt, + &agg.IndexedAt, + &recordURI, + &recordCID, + ) + if err != nil { + return nil, fmt.Errorf("failed to scan aggregator: %w", err) + } + + // Map nullable fields + agg.Description = description.String + agg.AvatarURL = avatarCID.String + agg.MaintainerDID = maintainerDID.String + agg.SourceURL = homepageURL.String + agg.RecordURI = recordURI.String + agg.RecordCID = recordCID.String + if configSchema != nil { + agg.ConfigSchema = configSchema + } + + aggs = append(aggs, agg) + } + + if err = rows.Err(); err != nil { + return nil, fmt.Errorf("error iterating aggregators: %w", err) + } + + return aggs, nil +} + +// IsAggregator performs a fast existence check for post creation handler +func (r *postgresAggregatorRepo) IsAggregator(ctx context.Context, did string) (bool, error) { + query := `SELECT EXISTS(SELECT 1 FROM aggregators WHERE did = $1)` + + var exists bool + err := r.db.QueryRowContext(ctx, query, did).Scan(&exists) + if err != nil { + return false, fmt.Errorf("failed to check if aggregator exists: %w", err) + } + + return exists, nil +} + +// ===== Authorization CRUD Operations ===== + +// CreateAuthorization indexes a new authorization from the firehose +func (r *postgresAggregatorRepo) CreateAuthorization(ctx context.Context, auth *aggregators.Authorization) error { + query := ` + INSERT INTO aggregator_authorizations ( + aggregator_did, community_did, enabled, config, + created_at, created_by, disabled_at, disabled_by, + indexed_at, record_uri, record_cid + ) VALUES ( + $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11 + ) + ON CONFLICT (aggregator_did, community_did) DO UPDATE SET + enabled = EXCLUDED.enabled, + config = EXCLUDED.config, + created_at = EXCLUDED.created_at, + created_by = EXCLUDED.created_by, + disabled_at = EXCLUDED.disabled_at, + disabled_by = EXCLUDED.disabled_by, + indexed_at = EXCLUDED.indexed_at, + record_uri = EXCLUDED.record_uri, + record_cid = EXCLUDED.record_cid + RETURNING id` + + var config interface{} + if len(auth.Config) > 0 { + config = auth.Config + } else { + config = nil + } + + var disabledAt interface{} + if auth.DisabledAt != nil { + disabledAt = *auth.DisabledAt + } else { + disabledAt = nil + } + + err := r.db.QueryRowContext(ctx, query, + auth.AggregatorDID, + auth.CommunityDID, + auth.Enabled, + config, + auth.CreatedAt, + auth.CreatedBy, // Required field, no nullString needed + disabledAt, + nullString(auth.DisabledBy), + auth.IndexedAt, + nullString(auth.RecordURI), + nullString(auth.RecordCID), + ).Scan(&auth.ID) + + if err != nil { + // Check for foreign key violations + if strings.Contains(err.Error(), "fk_aggregator") { + return aggregators.ErrAggregatorNotFound + } + return fmt.Errorf("failed to create authorization: %w", err) + } + + return nil +} + +// GetAuthorization retrieves an authorization by aggregator and community DID +func (r *postgresAggregatorRepo) GetAuthorization(ctx context.Context, aggregatorDID, communityDID string) (*aggregators.Authorization, error) { + query := ` + SELECT + id, aggregator_did, community_did, enabled, config, + created_at, created_by, disabled_at, disabled_by, + indexed_at, record_uri, record_cid + FROM aggregator_authorizations + WHERE aggregator_did = $1 AND community_did = $2` + + auth := &aggregators.Authorization{} + var config []byte + var createdBy, disabledBy, recordURI, recordCID sql.NullString + var disabledAt sql.NullTime + + err := r.db.QueryRowContext(ctx, query, aggregatorDID, communityDID).Scan( + &auth.ID, + &auth.AggregatorDID, + &auth.CommunityDID, + &auth.Enabled, + &config, + &auth.CreatedAt, + &createdBy, + &disabledAt, + &disabledBy, + &auth.IndexedAt, + &recordURI, + &recordCID, + ) + + if err == sql.ErrNoRows { + return nil, aggregators.ErrAuthorizationNotFound + } + if err != nil { + return nil, fmt.Errorf("failed to get authorization: %w", err) + } + + // Map nullable fields + auth.CreatedBy = createdBy.String + auth.DisabledBy = disabledBy.String + if disabledAt.Valid { + disabledAtVal := disabledAt.Time + auth.DisabledAt = &disabledAtVal + } + auth.RecordURI = recordURI.String + auth.RecordCID = recordCID.String + if config != nil { + auth.Config = config + } + + return auth, nil +} + +// GetAuthorizationByURI retrieves an authorization by record URI (for Jetstream delete operations) +func (r *postgresAggregatorRepo) GetAuthorizationByURI(ctx context.Context, recordURI string) (*aggregators.Authorization, error) { + query := ` + SELECT + id, aggregator_did, community_did, enabled, config, + created_at, created_by, disabled_at, disabled_by, + indexed_at, record_uri, record_cid + FROM aggregator_authorizations + WHERE record_uri = $1` + + auth := &aggregators.Authorization{} + var config []byte + var createdBy, disabledBy, recordURIField, recordCID sql.NullString + var disabledAt sql.NullTime + + err := r.db.QueryRowContext(ctx, query, recordURI).Scan( + &auth.ID, + &auth.AggregatorDID, + &auth.CommunityDID, + &auth.Enabled, + &config, + &auth.CreatedAt, + &createdBy, + &disabledAt, + &disabledBy, + &auth.IndexedAt, + &recordURIField, + &recordCID, + ) + + if err == sql.ErrNoRows { + return nil, aggregators.ErrAuthorizationNotFound + } + if err != nil { + return nil, fmt.Errorf("failed to get authorization by URI: %w", err) + } + + // Map nullable fields + auth.CreatedBy = createdBy.String + auth.DisabledBy = disabledBy.String + if disabledAt.Valid { + disabledAtVal := disabledAt.Time + auth.DisabledAt = &disabledAtVal + } + auth.RecordURI = recordURIField.String + auth.RecordCID = recordCID.String + if config != nil { + auth.Config = config + } + + return auth, nil +} + +// UpdateAuthorization updates an existing authorization +func (r *postgresAggregatorRepo) UpdateAuthorization(ctx context.Context, auth *aggregators.Authorization) error { + query := ` + UPDATE aggregator_authorizations SET + enabled = $3, + config = $4, + created_at = $5, + created_by = $6, + disabled_at = $7, + disabled_by = $8, + indexed_at = $9, + record_uri = $10, + record_cid = $11 + WHERE aggregator_did = $1 AND community_did = $2` + + var config interface{} + if len(auth.Config) > 0 { + config = auth.Config + } else { + config = nil + } + + var disabledAt interface{} + if auth.DisabledAt != nil { + disabledAt = *auth.DisabledAt + } else { + disabledAt = nil + } + + result, err := r.db.ExecContext(ctx, query, + auth.AggregatorDID, + auth.CommunityDID, + auth.Enabled, + config, + auth.CreatedAt, + nullString(auth.CreatedBy), + disabledAt, + nullString(auth.DisabledBy), + auth.IndexedAt, + nullString(auth.RecordURI), + nullString(auth.RecordCID), + ) + + if err != nil { + return fmt.Errorf("failed to update authorization: %w", err) + } + + rows, err := result.RowsAffected() + if err != nil { + return fmt.Errorf("failed to get rows affected: %w", err) + } + if rows == 0 { + return aggregators.ErrAuthorizationNotFound + } + + return nil +} + +// DeleteAuthorization removes an authorization +func (r *postgresAggregatorRepo) DeleteAuthorization(ctx context.Context, aggregatorDID, communityDID string) error { + query := `DELETE FROM aggregator_authorizations WHERE aggregator_did = $1 AND community_did = $2` + + result, err := r.db.ExecContext(ctx, query, aggregatorDID, communityDID) + if err != nil { + return fmt.Errorf("failed to delete authorization: %w", err) + } + + rows, err := result.RowsAffected() + if err != nil { + return fmt.Errorf("failed to get rows affected: %w", err) + } + if rows == 0 { + return aggregators.ErrAuthorizationNotFound + } + + return nil +} + +// DeleteAuthorizationByURI removes an authorization by record URI (for Jetstream delete operations) +func (r *postgresAggregatorRepo) DeleteAuthorizationByURI(ctx context.Context, recordURI string) error { + query := `DELETE FROM aggregator_authorizations WHERE record_uri = $1` + + result, err := r.db.ExecContext(ctx, query, recordURI) + if err != nil { + return fmt.Errorf("failed to delete authorization by URI: %w", err) + } + + rows, err := result.RowsAffected() + if err != nil { + return fmt.Errorf("failed to get rows affected: %w", err) + } + if rows == 0 { + return aggregators.ErrAuthorizationNotFound + } + + return nil +} + +// ===== Authorization Query Operations ===== + +// ListAuthorizationsForAggregator retrieves all communities that authorized an aggregator +func (r *postgresAggregatorRepo) ListAuthorizationsForAggregator(ctx context.Context, aggregatorDID string, enabledOnly bool, limit, offset int) ([]*aggregators.Authorization, error) { + baseQuery := ` + SELECT + id, aggregator_did, community_did, enabled, config, + created_at, created_by, disabled_at, disabled_by, + indexed_at, record_uri, record_cid + FROM aggregator_authorizations + WHERE aggregator_did = $1` + + var query string + var args []interface{} + + if enabledOnly { + query = baseQuery + ` AND enabled = true ORDER BY created_at DESC LIMIT $2 OFFSET $3` + args = []interface{}{aggregatorDID, limit, offset} + } else { + query = baseQuery + ` ORDER BY created_at DESC LIMIT $2 OFFSET $3` + args = []interface{}{aggregatorDID, limit, offset} + } + + rows, err := r.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, fmt.Errorf("failed to list authorizations for aggregator: %w", err) + } + defer rows.Close() + + return scanAuthorizations(rows) +} + +// ListAuthorizationsForCommunity retrieves all aggregators authorized by a community +func (r *postgresAggregatorRepo) ListAuthorizationsForCommunity(ctx context.Context, communityDID string, enabledOnly bool, limit, offset int) ([]*aggregators.Authorization, error) { + baseQuery := ` + SELECT + id, aggregator_did, community_did, enabled, config, + created_at, created_by, disabled_at, disabled_by, + indexed_at, record_uri, record_cid + FROM aggregator_authorizations + WHERE community_did = $1` + + var query string + var args []interface{} + + if enabledOnly { + query = baseQuery + ` AND enabled = true ORDER BY created_at DESC LIMIT $2 OFFSET $3` + args = []interface{}{communityDID, limit, offset} + } else { + query = baseQuery + ` ORDER BY created_at DESC LIMIT $2 OFFSET $3` + args = []interface{}{communityDID, limit, offset} + } + + rows, err := r.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, fmt.Errorf("failed to list authorizations for community: %w", err) + } + defer rows.Close() + + return scanAuthorizations(rows) +} + +// IsAuthorized performs a fast authorization check (enabled=true) +// Uses the optimized partial index: idx_aggregator_auth_enabled +func (r *postgresAggregatorRepo) IsAuthorized(ctx context.Context, aggregatorDID, communityDID string) (bool, error) { + query := ` + SELECT EXISTS( + SELECT 1 FROM aggregator_authorizations + WHERE aggregator_did = $1 AND community_did = $2 AND enabled = true + )` + + var authorized bool + err := r.db.QueryRowContext(ctx, query, aggregatorDID, communityDID).Scan(&authorized) + if err != nil { + return false, fmt.Errorf("failed to check authorization: %w", err) + } + + return authorized, nil +} + +// ===== Post Tracking Operations ===== + +// RecordAggregatorPost tracks a post created by an aggregator (for rate limiting and stats) +func (r *postgresAggregatorRepo) RecordAggregatorPost(ctx context.Context, aggregatorDID, communityDID, postURI, postCID string) error { + query := ` + INSERT INTO aggregator_posts (aggregator_did, community_did, post_uri, post_cid, created_at) + VALUES ($1, $2, $3, $4, NOW())` + + _, err := r.db.ExecContext(ctx, query, aggregatorDID, communityDID, postURI, postCID) + if err != nil { + return fmt.Errorf("failed to record aggregator post: %w", err) + } + + return nil +} + +// CountRecentPosts counts posts created by an aggregator in a community since a given time +// Uses the optimized index: idx_aggregator_posts_rate_limit +func (r *postgresAggregatorRepo) CountRecentPosts(ctx context.Context, aggregatorDID, communityDID string, since time.Time) (int, error) { + query := ` + SELECT COUNT(*) + FROM aggregator_posts + WHERE aggregator_did = $1 AND community_did = $2 AND created_at >= $3` + + var count int + err := r.db.QueryRowContext(ctx, query, aggregatorDID, communityDID, since).Scan(&count) + if err != nil { + return 0, fmt.Errorf("failed to count recent posts: %w", err) + } + + return count, nil +} + +// GetRecentPosts retrieves recent posts created by an aggregator in a community +func (r *postgresAggregatorRepo) GetRecentPosts(ctx context.Context, aggregatorDID, communityDID string, since time.Time) ([]*aggregators.AggregatorPost, error) { + query := ` + SELECT id, aggregator_did, community_did, post_uri, created_at + FROM aggregator_posts + WHERE aggregator_did = $1 AND community_did = $2 AND created_at >= $3 + ORDER BY created_at DESC` + + rows, err := r.db.QueryContext(ctx, query, aggregatorDID, communityDID, since) + if err != nil { + return nil, fmt.Errorf("failed to get recent posts: %w", err) + } + defer rows.Close() + + var posts []*aggregators.AggregatorPost + for rows.Next() { + post := &aggregators.AggregatorPost{} + err := rows.Scan( + &post.ID, + &post.AggregatorDID, + &post.CommunityDID, + &post.PostURI, + &post.CreatedAt, + ) + if err != nil { + return nil, fmt.Errorf("failed to scan aggregator post: %w", err) + } + posts = append(posts, post) + } + + if err = rows.Err(); err != nil { + return nil, fmt.Errorf("error iterating aggregator posts: %w", err) + } + + return posts, nil +} + +// ===== Helper Functions ===== + +// scanAuthorizations is a helper to scan multiple authorization rows +func scanAuthorizations(rows *sql.Rows) ([]*aggregators.Authorization, error) { + var auths []*aggregators.Authorization + + for rows.Next() { + auth := &aggregators.Authorization{} + var config []byte + var createdBy, disabledBy, recordURI, recordCID sql.NullString + var disabledAt sql.NullTime + + err := rows.Scan( + &auth.ID, + &auth.AggregatorDID, + &auth.CommunityDID, + &auth.Enabled, + &config, + &auth.CreatedAt, + &createdBy, + &disabledAt, + &disabledBy, + &auth.IndexedAt, + &recordURI, + &recordCID, + ) + if err != nil { + return nil, fmt.Errorf("failed to scan authorization: %w", err) + } + + // Map nullable fields + auth.CreatedBy = createdBy.String + auth.DisabledBy = disabledBy.String + if disabledAt.Valid { + disabledAtVal := disabledAt.Time + auth.DisabledAt = &disabledAtVal + } + auth.RecordURI = recordURI.String + auth.RecordCID = recordCID.String + if config != nil { + auth.Config = config + } + + auths = append(auths, auth) + } + + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("error iterating authorizations: %w", err) + } + + return auths, nil +}