feat(communities): add avatar and banner blob upload support

Implement full avatar and banner image support for community creation and updates:

- Add BlobOwner interface to break import cycle between blobs and communities
- Community and CommunityPDSAccount now implement BlobOwner for blob uploads
- CreateCommunity and UpdateCommunity now handle avatar/banner blob uploads to PDS
- Add avatarMimeType/bannerMimeType fields to request structs and lexicons
- Move authorization check before blob uploads to prevent orphaned blob DoS
- Add PDS response validation for blob uploads
- Fix Makefile db-migrate port to match .env.dev (5435)

Includes comprehensive E2E tests for avatar/banner CRUD operations via Jetstream.

Co-Authored-By: Claude Opus 4.5 <[e-mail]>

Makefile

···
 
 db-migrate: ## Run database migrations
 	@echo "$(GREEN)Running database migrations...$(RESET)"
-	@goose -dir internal/db/migrations postgres "postgresql://dev_user:dev_password@localhost:5433/coves_dev?sslmode=disable" up
+	@goose -dir internal/db/migrations postgres "postgresql://dev_user:dev_password@localhost:5435/coves_dev?sslmode=disable" up
 	@echo "$(GREEN)✓ Migrations complete$(RESET)"
 
 db-migrate-down: ## Rollback last migration
 	@echo "$(YELLOW)Rolling back last migration...$(RESET)"
-	@goose -dir internal/db/migrations postgres "postgresql://dev_user:dev_password@localhost:5433/coves_dev?sslmode=disable" down
+	@goose -dir internal/db/migrations postgres "postgresql://dev_user:dev_password@localhost:5435/coves_dev?sslmode=disable" down
 	@echo "$(GREEN)✓ Rollback complete$(RESET)"
 
 db-reset: ## Reset database (delete all data and re-run migrations)

cmd/server/main.go

···
 	log.Printf("   - Communities will be created at: %s", defaultPDS)
 	log.Printf("   - PDS will generate and manage all DIDs and keys")
 
+	// Initialize blob upload service (moved earlier for community service)
+	blobService := blobs.NewBlobService(defaultPDS)
+	log.Println("✅ Blob service initialized")
+
 	// Initialize community service with OAuth client for user DPoP authentication
 	// OAuth client is required for subscribe/unsubscribe/block/unblock operations
-	communityService := communities.NewCommunityService(communityRepo, defaultPDS, instanceDID, instanceDomain, provisioner, oauthClient)
+	communityService := communities.NewCommunityService(
+		communityRepo,
+		defaultPDS,
+		instanceDID,
+		instanceDomain,
+		provisioner,
+		oauthClient,
+		blobService,
+	)
 
 	// Authenticate Coves instance with PDS to enable community record writes
 	// The instance needs a PDS account to write community records it owns
···
 
 	// Initialize unfurl cache repository
 	unfurlRepo := unfurl.NewRepository(db)
-
-	// Initialize blob upload service
-	blobService := blobs.NewBlobService(defaultPDS)
 
 	// Initialize unfurl service with configuration
 	unfurlService := unfurl.NewService(

internal/atproto/lexicon/social/coves/community/create.json

···
               "accept": ["image/png", "image/jpeg", "image/webp"],
               "maxSize": 2000000
             },
+            "avatarMimeType": {
+              "type": "string",
+              "maxLength": 128,
+              "description": "MIME type of avatar blob (image/png, image/jpeg, image/webp)"
+            },
+            "bannerMimeType": {
+              "type": "string",
+              "maxLength": 128,
+              "description": "MIME type of banner blob (image/png, image/jpeg, image/webp)"
+            },
             "rules": {
               "type": "array",
               "maxLength": 10,
···
             },
             "handle": {
               "type": "string",
+              "maxLength": 512,
               "description": "Scoped handle of the created community (~name@instance)"
             }
           }

internal/atproto/lexicon/social/coves/community/update.json

···
               "accept": ["image/png", "image/jpeg", "image/webp"],
               "maxSize": 2000000
             },
+            "avatarMimeType": {
+              "type": "string",
+              "maxLength": 128,
+              "description": "MIME type of avatar blob (image/png, image/jpeg, image/webp)"
+            },
+            "bannerMimeType": {
+              "type": "string",
+              "maxLength": 128,
+              "description": "MIME type of banner blob (image/png, image/jpeg, image/webp)"
+            },
             "rules": {
               "type": "array",
               "maxLength": 10,

internal/core/blobs/service.go

···
 package blobs
 
 import (
-	"Coves/internal/core/communities"
 	"bytes"
 	"context"
 	"encoding/json"
···
 	"time"
 )
 
+// BlobOwner represents any entity that can own blobs on a PDS.
+// This interface breaks the import cycle between blobs and communities packages.
+// communities.Community implements this interface.
+type BlobOwner interface {
+	// GetPDSURL returns the PDS URL for this entity
+	GetPDSURL() string
+	// GetPDSAccessToken returns the access token for authenticating with the PDS
+	GetPDSAccessToken() string
+}
+
 // Service defines the interface for blob operations
 type Service interface {
-	// UploadBlobFromURL fetches an image from a URL and uploads it to the community's PDS
-	UploadBlobFromURL(ctx context.Context, community *communities.Community, imageURL string) (*BlobRef, error)
+	// UploadBlobFromURL fetches an image from a URL and uploads it to the owner's PDS
+	UploadBlobFromURL(ctx context.Context, owner BlobOwner, imageURL string) (*BlobRef, error)
 
-	// UploadBlob uploads binary data to the community's PDS
-	UploadBlob(ctx context.Context, community *communities.Community, data []byte, mimeType string) (*BlobRef, error)
+	// UploadBlob uploads binary data to the owner's PDS
+	UploadBlob(ctx context.Context, owner BlobOwner, data []byte, mimeType string) (*BlobRef, error)
 }
 
 type blobService struct {
···
 // UploadBlobFromURL fetches an image from a URL and uploads it to PDS
 // Flow:
 // 1. Fetch image from URL with timeout
-// 2. Validate size (<1MB)
+// 2. Validate size (max 6MB)
 // 3. Validate MIME type (image/jpeg, image/png, image/webp)
 // 4. Call UploadBlob to upload to PDS
-func (s *blobService) UploadBlobFromURL(ctx context.Context, community *communities.Community, imageURL string) (*BlobRef, error) {
+func (s *blobService) UploadBlobFromURL(ctx context.Context, owner BlobOwner, imageURL string) (*BlobRef, error) {
 	// Input validation
 	if imageURL == "" {
 		return nil, fmt.Errorf("image URL cannot be empty")
···
 	}
 
 	// Upload to PDS
-	return s.UploadBlob(ctx, community, data, mimeType)
+	return s.UploadBlob(ctx, owner, data, mimeType)
 }
 
-// UploadBlob uploads binary data to the community's PDS
+// UploadBlob uploads binary data to the owner's PDS
 // Flow:
 // 1. Validate inputs
 // 2. POST to {PDSURL}/xrpc/com.atproto.repo.uploadBlob
-// 3. Use community's PDSAccessToken for auth
+// 3. Use owner's PDSAccessToken for auth
 // 4. Set Content-Type header to mimeType
 // 5. Parse response and extract blob reference
-func (s *blobService) UploadBlob(ctx context.Context, community *communities.Community, data []byte, mimeType string) (*BlobRef, error) {
+func (s *blobService) UploadBlob(ctx context.Context, owner BlobOwner, data []byte, mimeType string) (*BlobRef, error) {
 	// Input validation
-	if community == nil {
-		return nil, fmt.Errorf("community cannot be nil")
+	if owner == nil {
+		return nil, fmt.Errorf("owner cannot be nil")
 	}
 	if len(data) == 0 {
 		return nil, fmt.Errorf("data cannot be empty")
···
 		return nil, fmt.Errorf("data size %d bytes exceeds maximum of %d bytes (6MB)", len(data), maxSize)
 	}
 
-	// Use community's PDS URL (for federated communities)
-	pdsURL := community.PDSURL
+	// Use owner's PDS URL (for federated communities)
+	pdsURL := owner.GetPDSURL()
 	if pdsURL == "" {
-		// Fallback to service default if community doesn't have a PDS URL
-		pdsURL = s.pdsURL
+		return nil, fmt.Errorf("owner has no PDS URL configured")
+	}
+
+	// Validate access token before making request
+	accessToken := owner.GetPDSAccessToken()
+	if accessToken == "" {
+		return nil, fmt.Errorf("owner has no PDS access token")
 	}
 
 	// Build PDS endpoint URL
···
 
 	// Set headers (auth + content type)
 	req.Header.Set("Content-Type", mimeType)
-	req.Header.Set("Authorization", "Bearer "+community.PDSAccessToken)
+	req.Header.Set("Authorization", "Bearer "+accessToken)
 
 	// Create HTTP client with timeout
 	client := &http.Client{
···
 	}
 	if err := json.Unmarshal(body, &result); err != nil {
 		return nil, fmt.Errorf("failed to parse PDS response: %w", err)
+	}
+
+	// Validate required fields in PDS response
+	if result.Blob.Type == "" {
+		return nil, fmt.Errorf("PDS response missing required field: $type")
+	}
+	if result.Blob.Ref == nil || result.Blob.Ref["$link"] == "" {
+		return nil, fmt.Errorf("PDS response missing required field: ref.$link (CID)")
+	}
+	if result.Blob.MimeType == "" {
+		return nil, fmt.Errorf("PDS response missing required field: mimeType")
+	}
+	if result.Blob.Size == 0 {
+		return nil, fmt.Errorf("PDS response missing required field: size")
 	}
 
 	return &result.Blob, nil

internal/core/communities/community.go

···
 	HostedByDID            string   `json:"hostedByDid"`
 	AvatarBlob             []byte   `json:"avatarBlob,omitempty"`
 	BannerBlob             []byte   `json:"bannerBlob,omitempty"`
+	AvatarMimeType         string   `json:"avatarMimeType,omitempty"`
+	BannerMimeType         string   `json:"bannerMimeType,omitempty"`
 	Rules                  []string `json:"rules,omitempty"`
 	Categories             []string `json:"categories,omitempty"`
 	AllowExternalDiscovery bool     `json:"allowExternalDiscovery"`
···
 	Description            *string  `json:"description,omitempty"`
 	AvatarBlob             []byte   `json:"avatarBlob,omitempty"`
 	BannerBlob             []byte   `json:"bannerBlob,omitempty"`
+	AvatarMimeType         string   `json:"avatarMimeType,omitempty"`
+	BannerMimeType         string   `json:"bannerMimeType,omitempty"`
 	Visibility             *string  `json:"visibility,omitempty"`
 	AllowExternalDiscovery *bool    `json:"allowExternalDiscovery,omitempty"`
 	ModerationType         *string  `json:"moderationType,omitempty"`
···
 
 	return fmt.Sprintf("!%s@%s", name, instanceDomain)
 }
+
+// GetPDSURL implements blobs.BlobOwner interface.
+// Returns the community's PDS URL for blob uploads.
+func (c *Community) GetPDSURL() string {
+	return c.PDSURL
+}
+
+// GetPDSAccessToken implements blobs.BlobOwner interface.
+// Returns the community's PDS access token for blob upload authentication.
+func (c *Community) GetPDSAccessToken() string {
+	return c.PDSAccessToken
+}

internal/core/communities/pds_provisioning.go

···
 	SigningKeyPEM  string // PEM-encoded signing key (for atproto operations)
 }
 
+// GetPDSURL implements blobs.BlobOwner interface.
+// Returns the community's PDS URL for blob uploads.
+func (c *CommunityPDSAccount) GetPDSURL() string {
+	return c.PDSURL
+}
+
+// GetPDSAccessToken implements blobs.BlobOwner interface.
+// Returns the community's PDS access token for blob upload authentication.
+func (c *CommunityPDSAccount) GetPDSAccessToken() string {
+	return c.AccessToken
+}
+
 // PDSAccountProvisioner creates PDS accounts for communities with PDS-managed DIDs
 type PDSAccountProvisioner struct {
 	instanceDomain string

internal/core/communities/service.go

···
 	oauthclient "Coves/internal/atproto/oauth"
 	"Coves/internal/atproto/pds"
 	"Coves/internal/atproto/utils"
+	"Coves/internal/core/blobs"
 	"bytes"
 	"context"
 	"encoding/json"
···
 	// Interfaces and pointers first (better alignment)
 	repo        Repository
 	provisioner *PDSAccountProvisioner
+	blobService blobs.Service
 
 	// OAuth client for user PDS authentication (DPoP-based)
 	oauthClient      *oauthclient.OAuthClient
···
 	pdsURL, instanceDID, instanceDomain string,
 	provisioner *PDSAccountProvisioner,
 	oauthClient *oauthclient.OAuthClient,
+	blobService blobs.Service,
 ) Service {
 	// SECURITY: Basic validation that did:web domain matches configured instanceDomain
 	// This catches honest configuration mistakes but NOT malicious code modifications
···
 		instanceDomain: instanceDomain,
 		provisioner:    provisioner,
 		oauthClient:    oauthClient,
+		blobService:    blobService,
 		refreshMutexes: make(map[string]*sync.Mutex),
 	}
 }
···
 	pdsURL, instanceDID, instanceDomain string,
 	provisioner *PDSAccountProvisioner,
 	factory PDSClientFactory,
+	blobService blobs.Service,
 ) Service {
 	return &communityService{
 		repo:             repo,
···
 		instanceDomain:   instanceDomain,
 		provisioner:      provisioner,
 		pdsClientFactory: factory,
+		blobService:      blobService,
 		refreshMutexes:   make(map[string]*sync.Mutex),
 	}
 }
···
 		profile["language"] = req.Language
 	}
 
-	// TODO: Handle avatar and banner blobs
-	// For now, we'll skip blob uploads. This would require:
-	// 1. Upload blob to PDS via com.atproto.repo.uploadBlob
-	// 2. Get blob ref (CID)
-	// 3. Add to profile record
+	// Track blob CIDs for storage in the community record
+	var avatarCID, bannerCID string
+
+	// Upload avatar if provided
+	if len(req.AvatarBlob) > 0 {
+		if req.AvatarMimeType == "" {
+			return nil, fmt.Errorf("avatarMimeType is required when avatarBlob is provided")
+		}
+		if s.blobService == nil {
+			return nil, fmt.Errorf("blob service not configured, cannot upload avatar")
+		}
+		avatarRef, err := s.blobService.UploadBlob(ctx, pdsAccount, req.AvatarBlob, req.AvatarMimeType)
+		if err != nil {
+			return nil, fmt.Errorf("failed to upload avatar: %w", err)
+		}
+		profile["avatar"] = map[string]interface{}{
+			"$type":    avatarRef.Type,
+			"ref":      avatarRef.Ref,
+			"mimeType": avatarRef.MimeType,
+			"size":     avatarRef.Size,
+		}
+		// Extract CID for database storage
+		avatarCID = avatarRef.Ref["$link"]
+	}
+
+	// Upload banner if provided
+	if len(req.BannerBlob) > 0 {
+		if req.BannerMimeType == "" {
+			return nil, fmt.Errorf("bannerMimeType is required when bannerBlob is provided")
+		}
+		if s.blobService == nil {
+			return nil, fmt.Errorf("blob service not configured, cannot upload banner")
+		}
+		bannerRef, err := s.blobService.UploadBlob(ctx, pdsAccount, req.BannerBlob, req.BannerMimeType)
+		if err != nil {
+			return nil, fmt.Errorf("failed to upload banner: %w", err)
+		}
+		profile["banner"] = map[string]interface{}{
+			"$type":    bannerRef.Type,
+			"ref":      bannerRef.Ref,
+			"mimeType": bannerRef.MimeType,
+			"size":     bannerRef.Size,
+		}
+		// Extract CID for database storage
+		bannerCID = bannerRef.Ref["$link"]
+	}
 
 	// V2: Write to COMMUNITY's own repository (not instance repo!)
 	// Repository: at://COMMUNITY_DID/social.coves.community.profile/self
···
 		PDSURL:                 pdsAccount.PDSURL,
 		Visibility:             req.Visibility,
 		AllowExternalDiscovery: req.AllowExternalDiscovery,
+		AvatarCID:              avatarCID,
+		BannerCID:              bannerCID,
 		MemberCount:            0,
 		SubscriberCount:        0,
 		CreatedAt:              time.Now(),
···
 		return nil, err
 	}
 
+	// Authorization: verify user is the creator BEFORE any expensive operations
+	// This prevents unauthorized users from uploading orphaned blobs (DoS vector)
+	// TODO(Communities-Auth): Add moderator check when moderation system is implemented
+	if existing.CreatedByDID != req.UpdatedByDID {
+		return nil, ErrUnauthorized
+	}
+
 	// CRITICAL: Ensure fresh PDS access token before write operation
 	// Community PDS tokens expire every ~2 hours and must be refreshed
 	existing, err = s.EnsureFreshToken(ctx, existing)
···
 		return nil, fmt.Errorf("failed to ensure fresh credentials: %w", err)
 	}
 
-	// Authorization: verify user is the creator
-	// TODO(Communities-Auth): Add moderator check when moderation system is implemented
-	if existing.CreatedByDID != req.UpdatedByDID {
-		return nil, ErrUnauthorized
+	// Upload avatar if provided
+	var avatarRef *blobs.BlobRef
+	if len(req.AvatarBlob) > 0 {
+		if req.AvatarMimeType == "" {
+			return nil, fmt.Errorf("avatarMimeType is required when avatarBlob is provided")
+		}
+		if s.blobService == nil {
+			return nil, fmt.Errorf("blob service not configured, cannot upload avatar")
+		}
+		ref, err := s.blobService.UploadBlob(ctx, existing, req.AvatarBlob, req.AvatarMimeType)
+		if err != nil {
+			return nil, fmt.Errorf("failed to upload avatar: %w", err)
+		}
+		avatarRef = ref
+	}
+
+	// Upload banner if provided
+	var bannerRef *blobs.BlobRef
+	if len(req.BannerBlob) > 0 {
+		if req.BannerMimeType == "" {
+			return nil, fmt.Errorf("bannerMimeType is required when bannerBlob is provided")
+		}
+		if s.blobService == nil {
+			return nil, fmt.Errorf("blob service not configured, cannot upload banner")
+		}
+		ref, err := s.blobService.UploadBlob(ctx, existing, req.BannerBlob, req.BannerMimeType)
+		if err != nil {
+			return nil, fmt.Errorf("failed to upload banner: %w", err)
+		}
+		bannerRef = ref
 	}
 
 	// Build updated profile record (start with existing)
···
 		profile["contentWarnings"] = req.ContentWarnings
 	} else if len(existing.ContentWarnings) > 0 {
 		profile["contentWarnings"] = existing.ContentWarnings
+	}
+
+	// Add blob references if uploaded
+	if avatarRef != nil {
+		profile["avatar"] = map[string]interface{}{
+			"$type":    avatarRef.Type,
+			"ref":      avatarRef.Ref,
+			"mimeType": avatarRef.MimeType,
+			"size":     avatarRef.Size,
+		}
+	}
+
+	if bannerRef != nil {
+		profile["banner"] = map[string]interface{}{
+			"$type":    bannerRef.Type,
+			"ref":      bannerRef.Ref,
+			"mimeType": bannerRef.MimeType,
+			"size":     bannerRef.Size,
+		}
 	}
 
 	// V2: Community profiles always use "self" as rkey

tests/integration/aggregator_e2e_test.go

···
 	identityConfig := identity.DefaultConfig()
 	identityResolver := identity.NewResolver(db, identityConfig)
 	userService := users.NewUserService(userRepo, identityResolver, "http://localhost:3001")
-	communityService := communities.NewCommunityServiceWithPDSFactory(communityRepo, "http://localhost:3001", "did:web:test.coves.social", "coves.social", nil, nil)
+	communityService := communities.NewCommunityServiceWithPDSFactory(communityRepo, "http://localhost:3001", "did:web:test.coves.social", "coves.social", nil, nil, nil)
 	aggregatorService := aggregators.NewAggregatorService(aggregatorRepo, communityService)
 	postService := posts.NewPostService(postRepo, communityService, aggregatorService, nil, nil, nil, "http://localhost:3001")
 

tests/integration/author_posts_e2e_test.go

···
 	// Setup services
 	resolver := identity.NewResolver(db, identity.DefaultConfig())
 	userService := users.NewUserService(userRepo, resolver, pdsURL)
-	communityService := communities.NewCommunityServiceWithPDSFactory(communityRepo, pdsURL, getTestInstanceDID(), "", nil, nil)
+	communityService := communities.NewCommunityServiceWithPDSFactory(communityRepo, pdsURL, getTestInstanceDID(), "", nil, nil, nil)
 	postService := posts.NewPostService(postRepo, communityService, nil, nil, nil, nil, pdsURL)
 	voteService := votes.NewServiceWithPDSFactory(voteRepo, nil, nil, PasswordAuthPDSClientFactory())
 
···
 
 	resolver := identity.NewResolver(db, identity.DefaultConfig())
 	userService := users.NewUserService(userRepo, resolver, getTestPDSURL())
-	communityService := communities.NewCommunityServiceWithPDSFactory(communityRepo, getTestPDSURL(), getTestInstanceDID(), "", nil, nil)
+	communityService := communities.NewCommunityServiceWithPDSFactory(communityRepo, getTestPDSURL(), getTestInstanceDID(), "", nil, nil, nil)
 	postService := posts.NewPostService(postRepo, communityService, nil, nil, nil, nil, getTestPDSURL())
 	voteService := votes.NewServiceWithPDSFactory(voteRepo, nil, nil, PasswordAuthPDSClientFactory())
 
···
 
 	resolver := identity.NewResolver(db, identity.DefaultConfig())
 	userService := users.NewUserService(userRepo, resolver, getTestPDSURL())
-	communityService := communities.NewCommunityServiceWithPDSFactory(communityRepo, getTestPDSURL(), getTestInstanceDID(), "", nil, nil)
+	communityService := communities.NewCommunityServiceWithPDSFactory(communityRepo, getTestPDSURL(), getTestInstanceDID(), "", nil, nil, nil)
 	postService := posts.NewPostService(postRepo, communityService, nil, nil, nil, nil, getTestPDSURL())
 	voteService := votes.NewServiceWithPDSFactory(voteRepo, nil, nil, PasswordAuthPDSClientFactory())
 
···
 	// Setup services
 	resolver := identity.NewResolver(db, identity.DefaultConfig())
 	userService := users.NewUserService(userRepo, resolver, pdsURL)
-	communityService := communities.NewCommunityServiceWithPDSFactory(communityRepo, pdsURL, getTestInstanceDID(), "", nil, nil)
+	communityService := communities.NewCommunityServiceWithPDSFactory(communityRepo, pdsURL, getTestInstanceDID(), "", nil, nil, nil)
 	postService := posts.NewPostService(postRepo, communityService, nil, nil, nil, nil, pdsURL)
 	voteService := votes.NewServiceWithPDSFactory(voteRepo, nil, nil, PasswordAuthPDSClientFactory())
 
···
 
 	resolver := identity.NewResolver(db, identity.DefaultConfig())
 	userService := users.NewUserService(userRepo, resolver, getTestPDSURL())
-	communityService := communities.NewCommunityServiceWithPDSFactory(communityRepo, getTestPDSURL(), getTestInstanceDID(), "", nil, nil)
+	communityService := communities.NewCommunityServiceWithPDSFactory(communityRepo, getTestPDSURL(), getTestInstanceDID(), "", nil, nil, nil)
 	postService := posts.NewPostService(postRepo, communityService, nil, nil, nil, nil, getTestPDSURL())
 	voteService := votes.NewServiceWithPDSFactory(voteRepo, nil, nil, PasswordAuthPDSClientFactory())
 

tests/integration/block_handle_resolution_test.go

···
 		"coves.social",
 		nil, // No PDS HTTP client for this test
 		nil, // No PDS factory needed for this test
+		nil, // No blob service for this test
 	)
 
 	blockHandler := community.NewBlockHandler(communityService)
···
 		"coves.social",
 		nil,
 		nil, // No PDS factory needed for this test
+		nil, // No blob service for this test
 	)
 
 	blockHandler := community.NewBlockHandler(communityService)

tests/integration/community_avatar_e2e_test.go

···
+package integration
+
+import (
+	"Coves/internal/atproto/identity"
+	"Coves/internal/atproto/jetstream"
+	"Coves/internal/core/blobs"
+	"Coves/internal/core/communities"
+	"Coves/internal/db/postgres"
+	"bytes"
+	"context"
+	"database/sql"
+	"fmt"
+	"image"
+	"image/color"
+	"image/png"
+	"net/http"
+	"os"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/gorilla/websocket"
+	_ "github.com/lib/pq"
+	"github.com/pressly/goose/v3"
+)
+
+// createTestPNGImage creates a simple PNG image for testing
+// Panics on encoding error since this is a test helper and encoding should never fail
+// for simple in-memory images
+func createTestPNGImage(width, height int, c color.Color) []byte {
+	img := image.NewRGBA(image.Rect(0, 0, width, height))
+	for y := 0; y < height; y++ {
+		for x := 0; x < width; x++ {
+			img.Set(x, y, c)
+		}
+	}
+	var buf bytes.Buffer
+	if err := png.Encode(&buf, img); err != nil {
+		panic(fmt.Sprintf("createTestPNGImage: failed to encode PNG: %v", err))
+	}
+	return buf.Bytes()
+}
+
+// TestCommunityAvatarE2E_CreateWithAvatar tests creating a community with an avatar
+// Flow: CreateCommunity(avatar) → PDS uploadBlob + putRecord → Jetstream → Consumer → AppView
+func TestCommunityAvatarE2E_CreateWithAvatar(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping E2E test in short mode")
+	}
+
+	// Setup test database
+	dbURL := os.Getenv("TEST_DATABASE_URL")
+	if dbURL == "" {
+		dbURL = "postgres://test_user:test_password@localhost:5434/coves_test?sslmode=disable"
+	}
+
+	db, err := sql.Open("postgres", dbURL)
+	if err != nil {
+		t.Fatalf("Failed to connect to test database: %v", err)
+	}
+	defer func() { _ = db.Close() }()
+
+	// Run migrations
+	if dialectErr := goose.SetDialect("postgres"); dialectErr != nil {
+		t.Fatalf("Failed to set goose dialect: %v", dialectErr)
+	}
+	if migrateErr := goose.Up(db, "../../internal/db/migrations"); migrateErr != nil {
+		t.Fatalf("Failed to run migrations: %v", migrateErr)
+	}
+
+	// Check if PDS is running
+	pdsURL := os.Getenv("PDS_URL")
+	if pdsURL == "" {
+		pdsURL = "http://localhost:3001"
+	}
+
+	healthResp, err := http.Get(pdsURL + "/xrpc/_health")
+	if err != nil {
+		t.Skipf("PDS not running at %s: %v. Run 'make dev-up' to start.", pdsURL, err)
+	}
+	_ = healthResp.Body.Close()
+
+	// Check if Jetstream is running
+	pdsHostname := strings.TrimPrefix(pdsURL, "http://")
+	pdsHostname = strings.TrimPrefix(pdsHostname, "https://")
+	pdsHostname = strings.Split(pdsHostname, ":")[0]
+	jetstreamURL := fmt.Sprintf("ws://%s:6008/subscribe?wantedCollections=social.coves.community.profile", pdsHostname)
+
+	testConn, _, connErr := websocket.DefaultDialer.Dial(jetstreamURL, nil)
+	if connErr != nil {
+		t.Skipf("Jetstream not available at %s: %v. Run 'make dev-up' to start.", jetstreamURL, connErr)
+	}
+	_ = testConn.Close()
+	t.Logf("✅ Jetstream available at %s", jetstreamURL)
+
+	ctx := context.Background()
+	instanceDID := "did:web:coves.social"
+
+	// Setup identity resolver with local PLC
+	plcURL := os.Getenv("PLC_DIRECTORY_URL")
+	if plcURL == "" {
+		plcURL = "http://localhost:3002"
+	}
+	identityConfig := identity.DefaultConfig()
+	identityConfig.PLCURL = plcURL
+	identityResolver := identity.NewResolver(db, identityConfig)
+
+	// Setup services
+	communityRepo := postgres.NewCommunityRepository(db)
+	provisioner := communities.NewPDSAccountProvisioner("coves.social", pdsURL)
+	blobService := blobs.NewBlobService(pdsURL)
+
+	communityService := communities.NewCommunityServiceWithPDSFactory(
+		communityRepo,
+		pdsURL,
+		instanceDID,
+		"coves.social",
+		provisioner,
+		nil, // No custom PDS factory, uses built-in
+		blobService,
+	)
+
+	consumer := jetstream.NewCommunityEventConsumer(communityRepo, instanceDID, true, identityResolver)
+
+	t.Run("create community with avatar via real Jetstream", func(t *testing.T) {
+		uniqueName := fmt.Sprintf("avt%d", time.Now().UnixNano()%1000000)
+		creatorDID := "did:plc:avatar-create-test"
+
+		// Create a test PNG image (100x100 red square)
+		avatarData := createTestPNGImage(100, 100, color.RGBA{255, 0, 0, 255})
+		t.Logf("Created test avatar image: %d bytes", len(avatarData))
+
+		// Subscribe to Jetstream BEFORE creating the community
+		// This ensures we catch the create event
+		eventChan := make(chan *jetstream.JetstreamEvent, 10)
+		done := make(chan bool)
+		subscribeCtx, cancelSubscribe := context.WithTimeout(ctx, 30*time.Second)
+		defer cancelSubscribe()
+
+		// We don't know the DID yet, so we'll filter by collection and match after
+		go func() {
+			conn, _, dialErr := websocket.DefaultDialer.Dial(jetstreamURL, nil)
+			if dialErr != nil {
+				t.Logf("Failed to connect to Jetstream: %v", dialErr)
+				return
+			}
+			defer func() { _ = conn.Close() }()
+
+			for {
+				select {
+				case <-done:
+					return
+				case <-subscribeCtx.Done():
+					return
+				default:
+					if deadlineErr := conn.SetReadDeadline(time.Now().Add(5 * time.Second)); deadlineErr != nil {
+						return
+					}
+
+					var event jetstream.JetstreamEvent
+					if readErr := conn.ReadJSON(&event); readErr != nil {
+						continue // Timeout or error, keep trying
+					}
+
+					// Only process community profile create events
+					if event.Kind == "commit" && event.Commit != nil &&
+						event.Commit.Collection == "social.coves.community.profile" &&
+						event.Commit.Operation == "create" {
+						eventChan <- &event
+					}
+				}
+			}
+		}()
+		time.Sleep(500 * time.Millisecond) // Give subscriber time to connect
+
+		t.Logf("\n📝 Creating community with avatar on PDS...")
+		community, createErr := communityService.CreateCommunity(ctx, communities.CreateCommunityRequest{
+			Name:                   uniqueName,
+			DisplayName:            "Community With Avatar",
+			Description:            "Testing avatar upload on create",
+			Visibility:             "public",
+			CreatedByDID:           creatorDID,
+			HostedByDID:            instanceDID,
+			AllowExternalDiscovery: true,
+			AvatarBlob:             avatarData,
+			AvatarMimeType:         "image/png",
+		})
+		if createErr != nil {
+			close(done)
+			t.Fatalf("Failed to create community with avatar: %v", createErr)
+		}
+
+		t.Logf("✅ Community created on PDS:")
+		t.Logf("   DID: %s", community.DID)
+		t.Logf("   RecordCID: %s", community.RecordCID)
+		t.Logf("   AvatarCID (from service): %s", community.AvatarCID)
+
+		// Wait for REAL Jetstream event
+		t.Logf("\n⏳ Waiting for create event from Jetstream...")
+		var realEvent *jetstream.JetstreamEvent
+		timeout := time.After(15 * time.Second)
+
+	eventLoop:
+		for {
+			select {
+			case event := <-eventChan:
+				// Match by DID (we now know it)
+				if event.Did == community.DID {
+					realEvent = event
+					t.Logf("✅ Received REAL create event from Jetstream!")
+					t.Logf("   DID: %s", event.Did)
+					t.Logf("   Operation: %s", event.Commit.Operation)
+					t.Logf("   CID: %s", event.Commit.CID)
+
+					// Log avatar info from real event
+					if event.Commit.Record != nil {
+						if avatar, hasAvatar := event.Commit.Record["avatar"]; hasAvatar {
+							t.Logf("   Avatar in event: %v", avatar)
+						}
+					}
+					break eventLoop
+				}
+			case <-timeout:
+				close(done)
+				t.Fatalf("Timeout waiting for Jetstream create event for DID %s", community.DID)
+			}
+		}
+		close(done)
+
+		// Process the REAL event through consumer
+		// Note: The community already exists (service indexed it), so consumer will hit conflict
+		// But this tests that the real event has correct avatar data
+		t.Logf("\n🔄 Processing real Jetstream event through consumer...")
+		if handleErr := consumer.HandleEvent(ctx, realEvent); handleErr != nil {
+			t.Logf("   Note: Consumer conflict expected (already indexed): %v", handleErr)
+		}
+
+		// Verify avatar CID matches what's in the database
+		final, err := communityRepo.GetByDID(ctx, community.DID)
+		if err != nil {
+			t.Fatalf("Failed to get final community: %v", err)
+		}
+
+		t.Logf("\n✅ Community avatar verification:")
+		t.Logf("   AvatarCID in DB: %s", final.AvatarCID)
+
+		if final.AvatarCID == "" {
+			t.Errorf("Expected AvatarCID to be set after create with avatar")
+		}
+
+		// Verify the avatar CID from the real Jetstream event matches what we stored
+		if realEvent.Commit.Record != nil {
+			if avatar, hasAvatar := realEvent.Commit.Record["avatar"].(map[string]interface{}); hasAvatar {
+				if ref, hasRef := avatar["ref"].(map[string]interface{}); hasRef {
+					if link, hasLink := ref["$link"].(string); hasLink {
+						t.Logf("   AvatarCID from Jetstream: %s", link)
+						if final.AvatarCID != link {
+							t.Errorf("AvatarCID mismatch: DB has %s, Jetstream has %s", final.AvatarCID, link)
+						} else {
+							t.Logf("   ✅ AvatarCID matches between DB and Jetstream event!")
+						}
+					}
+				}
+			}
+		}
+
+		// Verify we can fetch the avatar from PDS
+		pdsResp, pdsErr := http.Get(fmt.Sprintf("%s/xrpc/com.atproto.repo.getRecord?repo=%s&collection=social.coves.community.profile&rkey=self",
+			pdsURL, community.DID))
+		if pdsErr != nil {
+			t.Fatalf("Failed to fetch profile record from PDS: %v", pdsErr)
+		}
+		defer func() { _ = pdsResp.Body.Close() }()
+
+		if pdsResp.StatusCode != http.StatusOK {
+			t.Fatalf("Profile record not found on PDS: status %d", pdsResp.StatusCode)
+		}
+		t.Logf("   ✅ Profile record with avatar exists on PDS")
+
+		t.Logf("\n✅ TRUE E2E AVATAR CREATE FLOW COMPLETE:")
+		t.Logf("   Service → PDS uploadBlob → PDS putRecord → Jetstream → Verified ✓")
+	})
+}
+
+// TestCommunityAvatarE2E_UpdateWithAvatar tests updating a community's avatar
+// Flow: UpdateCommunity(avatar) → PDS uploadBlob + putRecord → Jetstream → Consumer → AppView
+func TestCommunityAvatarE2E_UpdateWithAvatar(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping E2E test in short mode")
+	}
+
+	// Setup test database
+	dbURL := os.Getenv("TEST_DATABASE_URL")
+	if dbURL == "" {
+		dbURL = "postgres://test_user:test_password@localhost:5434/coves_test?sslmode=disable"
+	}
+
+	db, err := sql.Open("postgres", dbURL)
+	if err != nil {
+		t.Fatalf("Failed to connect to test database: %v", err)
+	}
+	defer func() { _ = db.Close() }()
+
+	// Run migrations
+	if dialectErr := goose.SetDialect("postgres"); dialectErr != nil {
+		t.Fatalf("Failed to set goose dialect: %v", dialectErr)
+	}
+	if migrateErr := goose.Up(db, "../../internal/db/migrations"); migrateErr != nil {
+		t.Fatalf("Failed to run migrations: %v", migrateErr)
+	}
+
+	// Check if PDS is running
+	pdsURL := os.Getenv("PDS_URL")
+	if pdsURL == "" {
+		pdsURL = "http://localhost:3001"
+	}
+
+	healthResp, err := http.Get(pdsURL + "/xrpc/_health")
+	if err != nil {
+		t.Skipf("PDS not running at %s: %v. Run 'make dev-up' to start.", pdsURL, err)
+	}
+	_ = healthResp.Body.Close()
+
+	// Check if Jetstream is running - REQUIRED for true E2E
+	pdsHostname := strings.TrimPrefix(pdsURL, "http://")
+	pdsHostname = strings.TrimPrefix(pdsHostname, "https://")
+	pdsHostname = strings.Split(pdsHostname, ":")[0]
+	jetstreamURL := fmt.Sprintf("ws://%s:6008/subscribe?wantedCollections=social.coves.community.profile", pdsHostname)
+
+	testConn, _, connErr := websocket.DefaultDialer.Dial(jetstreamURL, nil)
+	if connErr != nil {
+		t.Skipf("Jetstream not available at %s: %v. Run 'make dev-up' to start.", jetstreamURL, connErr)
+	}
+	_ = testConn.Close()
+	t.Logf("✅ Jetstream available at %s", jetstreamURL)
+
+	ctx := context.Background()
+	instanceDID := "did:web:coves.social"
+
+	// Setup identity resolver with local PLC
+	plcURL := os.Getenv("PLC_DIRECTORY_URL")
+	if plcURL == "" {
+		plcURL = "http://localhost:3002"
+	}
+	identityConfig := identity.DefaultConfig()
+	identityConfig.PLCURL = plcURL
+	identityResolver := identity.NewResolver(db, identityConfig)
+
+	// Setup services
+	communityRepo := postgres.NewCommunityRepository(db)
+	provisioner := communities.NewPDSAccountProvisioner("coves.social", pdsURL)
+	blobService := blobs.NewBlobService(pdsURL)
+
+	communityService := communities.NewCommunityServiceWithPDSFactory(
+		communityRepo,
+		pdsURL,
+		instanceDID,
+		"coves.social",
+		provisioner,
+		nil,
+		blobService,
+	)
+
+	consumer := jetstream.NewCommunityEventConsumer(communityRepo, instanceDID, true, identityResolver)
+
+	// Helper to wait for Jetstream update event and process it
+	waitForUpdateEvent := func(t *testing.T, communityDID string, timeout time.Duration) *jetstream.JetstreamEvent {
+		eventChan := make(chan *jetstream.JetstreamEvent, 10)
+		done := make(chan bool)
+		subscribeCtx, cancelSubscribe := context.WithTimeout(ctx, timeout)
+		defer cancelSubscribe()
+
+		go func() {
+			conn, _, dialErr := websocket.DefaultDialer.Dial(jetstreamURL, nil)
+			if dialErr != nil {
+				t.Logf("Failed to connect to Jetstream: %v", dialErr)
+				return
+			}
+			defer func() { _ = conn.Close() }()
+
+			for {
+				select {
+				case <-done:
+					return
+				case <-subscribeCtx.Done():
+					return
+				default:
+					if deadlineErr := conn.SetReadDeadline(time.Now().Add(5 * time.Second)); deadlineErr != nil {
+						return
+					}
+
+					var event jetstream.JetstreamEvent
+					if readErr := conn.ReadJSON(&event); readErr != nil {
+						continue
+					}
+
+					if event.Kind == "commit" && event.Commit != nil &&
+						event.Commit.Collection == "social.coves.community.profile" &&
+						event.Commit.Operation == "update" &&
+						event.Did == communityDID {
+						eventChan <- &event
+					}
+				}
+			}
+		}()
+
+		select {
+		case event := <-eventChan:
+			close(done)
+			return event
+		case <-time.After(timeout):
+			close(done)
+			return nil
+		}
+	}
+
+	t.Run("add avatar to community without one", func(t *testing.T) {
+		uniqueName := fmt.Sprintf("upav%d", time.Now().UnixNano()%1000000)
+		creatorDID := "did:plc:avatar-update-test"
+
+		// Create a community WITHOUT an avatar
+		t.Logf("\n📝 Creating community without avatar...")
+		community, createErr := communityService.CreateCommunity(ctx, communities.CreateCommunityRequest{
+			Name:                   uniqueName,
+			DisplayName:            "Community Without Avatar",
+			Description:            "Will add avatar via update",
+			Visibility:             "public",
+			CreatedByDID:           creatorDID,
+			HostedByDID:            instanceDID,
+			AllowExternalDiscovery: true,
+		})
+		if createErr != nil {
+			t.Fatalf("Failed to create community: %v", createErr)
+		}
+		t.Logf("✅ Community created: DID=%s", community.DID)
+
+		// Verify no avatar initially
+		initial, err := communityService.GetCommunity(ctx, community.DID)
+		if err != nil {
+			t.Fatalf("Community not indexed: %v", err)
+		}
+		if initial.AvatarCID != "" {
+			t.Fatalf("Expected no initial avatar, got: %s", initial.AvatarCID)
+		}
+		t.Logf("   Initial AvatarCID: '' (confirmed empty)")
+
+		// Create test avatar image (100x100 blue square)
+		avatarData := createTestPNGImage(100, 100, color.RGBA{0, 0, 255, 255})
+		t.Logf("\n📝 Updating community with avatar (%d bytes)...", len(avatarData))
+
+		// Start listening for Jetstream event
+		eventReceived := make(chan *jetstream.JetstreamEvent, 1)
+		go func() {
+			event := waitForUpdateEvent(t, community.DID, 15*time.Second)
+			eventReceived <- event
+		}()
+		time.Sleep(500 * time.Millisecond) // Give subscriber time to connect
+
+		// Perform the update with avatar
+		newDisplayName := "Community With New Avatar"
+		updated, updateErr := communityService.UpdateCommunity(ctx, communities.UpdateCommunityRequest{
+			CommunityDID:   community.DID,
+			UpdatedByDID:   creatorDID,
+			DisplayName:    &newDisplayName,
+			AvatarBlob:     avatarData,
+			AvatarMimeType: "image/png",
+		})
+		if updateErr != nil {
+			t.Fatalf("Failed to update community with avatar: %v", updateErr)
+		}
+
+		t.Logf("✅ Community update written to PDS:")
+		t.Logf("   New RecordCID: %s", updated.RecordCID)
+
+		// Wait for REAL Jetstream event
+		t.Logf("\n⏳ Waiting for update event from Jetstream...")
+		realEvent := <-eventReceived
+		if realEvent == nil {
+			t.Fatalf("Timeout waiting for Jetstream update event")
+		}
+
+		t.Logf("✅ Received REAL update event from Jetstream!")
+		t.Logf("   Operation: %s", realEvent.Commit.Operation)
+		t.Logf("   CID: %s", realEvent.Commit.CID)
+
+		// Extract avatar CID from real event
+		var avatarCIDFromEvent string
+		if realEvent.Commit.Record != nil {
+			if avatar, hasAvatar := realEvent.Commit.Record["avatar"].(map[string]interface{}); hasAvatar {
+				t.Logf("   Avatar in event: %v", avatar)
+				if ref, hasRef := avatar["ref"].(map[string]interface{}); hasRef {
+					if link, hasLink := ref["$link"].(string); hasLink {
+						avatarCIDFromEvent = link
+						t.Logf("   AvatarCID from Jetstream: %s", avatarCIDFromEvent)
+					}
+				}
+			}
+		}
+
+		// Process the REAL event through consumer
+		t.Logf("\n🔄 Processing real Jetstream event through consumer...")
+		if handleErr := consumer.HandleEvent(ctx, realEvent); handleErr != nil {
+			t.Logf("   Consumer error: %v", handleErr)
+		}
+
+		// Verify avatar CID is now set in DB
+		final, err := communityRepo.GetByDID(ctx, community.DID)
+		if err != nil {
+			t.Fatalf("Failed to get final community: %v", err)
+		}
+
+		t.Logf("\n✅ Community avatar update verified:")
+		t.Logf("   DisplayName: %s", final.DisplayName)
+		t.Logf("   AvatarCID in DB: %s", final.AvatarCID)
+
+		if final.AvatarCID == "" {
+			t.Errorf("Expected AvatarCID to be set after update")
+		}
+
+		// Verify DB matches Jetstream event
+		if avatarCIDFromEvent != "" && final.AvatarCID != avatarCIDFromEvent {
+			t.Errorf("AvatarCID mismatch: DB has %s, Jetstream has %s", final.AvatarCID, avatarCIDFromEvent)
+		} else if avatarCIDFromEvent != "" {
+			t.Logf("   ✅ AvatarCID matches between DB and Jetstream!")
+		}
+
+		t.Logf("\n✅ TRUE E2E ADD AVATAR FLOW COMPLETE")
+	})
+
+	t.Run("replace existing avatar with new one", func(t *testing.T) {
+		uniqueName := fmt.Sprintf("rpav%d", time.Now().UnixNano()%1000000)
+		creatorDID := "did:plc:avatar-replace-test"
+
+		// Create a community WITH an initial avatar (red square)
+		initialAvatarData := createTestPNGImage(100, 100, color.RGBA{255, 0, 0, 255})
+		t.Logf("\n📝 Creating community with initial avatar (red, %d bytes)...", len(initialAvatarData))
+
+		community, createErr := communityService.CreateCommunity(ctx, communities.CreateCommunityRequest{
+			Name:                   uniqueName,
+			DisplayName:            "Community With Initial Avatar",
+			Description:            "Will replace avatar",
+			Visibility:             "public",
+			CreatedByDID:           creatorDID,
+			HostedByDID:            instanceDID,
+			AllowExternalDiscovery: true,
+			AvatarBlob:             initialAvatarData,
+			AvatarMimeType:         "image/png",
+		})
+		if createErr != nil {
+			t.Fatalf("Failed to create community with avatar: %v", createErr)
+		}
+		t.Logf("✅ Community created: DID=%s", community.DID)
+
+		// Verify initial avatar is set
+		initial, err := communityService.GetCommunity(ctx, community.DID)
+		if err != nil {
+			t.Fatalf("Community not indexed: %v", err)
+		}
+		initialAvatarCID := initial.AvatarCID
+		if initialAvatarCID == "" {
+			t.Fatalf("Expected initial avatar to be set")
+		}
+		t.Logf("   Initial AvatarCID: %s", initialAvatarCID)
+
+		// Create NEW avatar image (100x100 green square - different from initial red)
+		newAvatarData := createTestPNGImage(100, 100, color.RGBA{0, 255, 0, 255})
+		t.Logf("\n📝 Replacing avatar with new one (green, %d bytes)...", len(newAvatarData))
+
+		// Start listening for Jetstream event
+		eventReceived := make(chan *jetstream.JetstreamEvent, 1)
+		go func() {
+			event := waitForUpdateEvent(t, community.DID, 15*time.Second)
+			eventReceived <- event
+		}()
+		time.Sleep(500 * time.Millisecond)
+
+		// Perform the update with NEW avatar
+		newDisplayName := "Community With Replaced Avatar"
+		updated, updateErr := communityService.UpdateCommunity(ctx, communities.UpdateCommunityRequest{
+			CommunityDID:   community.DID,
+			UpdatedByDID:   creatorDID,
+			DisplayName:    &newDisplayName,
+			AvatarBlob:     newAvatarData,
+			AvatarMimeType: "image/png",
+		})
+		if updateErr != nil {
+			t.Fatalf("Failed to update community with new avatar: %v", updateErr)
+		}
+
+		t.Logf("✅ Community update written to PDS:")
+		t.Logf("   New RecordCID: %s", updated.RecordCID)
+
+		// Wait for REAL Jetstream event
+		t.Logf("\n⏳ Waiting for update event from Jetstream...")
+		realEvent := <-eventReceived
+		if realEvent == nil {
+			t.Fatalf("Timeout waiting for Jetstream update event")
+		}
+
+		t.Logf("✅ Received REAL update event from Jetstream!")
+		t.Logf("   Operation: %s", realEvent.Commit.Operation)
+
+		// Extract new avatar CID from real event
+		var newAvatarCIDFromEvent string
+		if realEvent.Commit.Record != nil {
+			if avatar, hasAvatar := realEvent.Commit.Record["avatar"].(map[string]interface{}); hasAvatar {
+				if ref, hasRef := avatar["ref"].(map[string]interface{}); hasRef {
+					if link, hasLink := ref["$link"].(string); hasLink {
+						newAvatarCIDFromEvent = link
+						t.Logf("   New AvatarCID from Jetstream: %s", newAvatarCIDFromEvent)
+					}
+				}
+			}
+		}
+
+		// Process the REAL event through consumer
+		t.Logf("\n🔄 Processing real Jetstream event through consumer...")
+		if handleErr := consumer.HandleEvent(ctx, realEvent); handleErr != nil {
+			t.Logf("   Consumer error: %v", handleErr)
+		}
+
+		// Verify avatar CID has CHANGED
+		final, err := communityRepo.GetByDID(ctx, community.DID)
+		if err != nil {
+			t.Fatalf("Failed to get final community: %v", err)
+		}
+
+		t.Logf("\n✅ Community avatar replacement verified:")
+		t.Logf("   DisplayName: %s", final.DisplayName)
+		t.Logf("   Old AvatarCID: %s", initialAvatarCID)
+		t.Logf("   New AvatarCID: %s", final.AvatarCID)
+
+		if final.AvatarCID == "" {
+			t.Errorf("Expected AvatarCID to be set after replacement")
+		}
+
+		if final.AvatarCID == initialAvatarCID {
+			t.Errorf("AvatarCID should have changed after replacement! Old: %s, New: %s", initialAvatarCID, final.AvatarCID)
+		} else {
+			t.Logf("   ✅ AvatarCID successfully changed!")
+		}
+
+		// Verify DB matches Jetstream event
+		if newAvatarCIDFromEvent != "" && final.AvatarCID != newAvatarCIDFromEvent {
+			t.Errorf("AvatarCID mismatch: DB has %s, Jetstream has %s", final.AvatarCID, newAvatarCIDFromEvent)
+		} else if newAvatarCIDFromEvent != "" {
+			t.Logf("   ✅ New AvatarCID matches between DB and Jetstream!")
+		}
+
+		t.Logf("\n✅ TRUE E2E REPLACE AVATAR FLOW COMPLETE")
+	})
+}
+
+// TestCommunityAvatarE2E_UpdateWithBanner tests updating a community's banner
+// Flow: UpdateCommunity(banner) → PDS uploadBlob + putRecord → Jetstream → Consumer → AppView
+func TestCommunityAvatarE2E_UpdateWithBanner(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping E2E test in short mode")
+	}
+
+	// Setup test database
+	dbURL := os.Getenv("TEST_DATABASE_URL")
+	if dbURL == "" {
+		dbURL = "postgres://test_user:test_password@localhost:5434/coves_test?sslmode=disable"
+	}
+
+	db, err := sql.Open("postgres", dbURL)
+	if err != nil {
+		t.Fatalf("Failed to connect to test database: %v", err)
+	}
+	defer func() { _ = db.Close() }()
+
+	// Run migrations
+	if dialectErr := goose.SetDialect("postgres"); dialectErr != nil {
+		t.Fatalf("Failed to set goose dialect: %v", dialectErr)
+	}
+	if migrateErr := goose.Up(db, "../../internal/db/migrations"); migrateErr != nil {
+		t.Fatalf("Failed to run migrations: %v", migrateErr)
+	}
+
+	// Check if PDS is running
+	pdsURL := os.Getenv("PDS_URL")
+	if pdsURL == "" {
+		pdsURL = "http://localhost:3001"
+	}
+
+	healthResp, err := http.Get(pdsURL + "/xrpc/_health")
+	if err != nil {
+		t.Skipf("PDS not running at %s: %v. Run 'make dev-up' to start.", pdsURL, err)
+	}
+	_ = healthResp.Body.Close()
+
+	// Check if Jetstream is running - REQUIRED for true E2E
+	pdsHostname := strings.TrimPrefix(pdsURL, "http://")
+	pdsHostname = strings.TrimPrefix(pdsHostname, "https://")
+	pdsHostname = strings.Split(pdsHostname, ":")[0]
+	jetstreamURL := fmt.Sprintf("ws://%s:6008/subscribe?wantedCollections=social.coves.community.profile", pdsHostname)
+
+	testConn, _, connErr := websocket.DefaultDialer.Dial(jetstreamURL, nil)
+	if connErr != nil {
+		t.Skipf("Jetstream not available at %s: %v. Run 'make dev-up' to start.", jetstreamURL, connErr)
+	}
+	_ = testConn.Close()
+	t.Logf("✅ Jetstream available at %s", jetstreamURL)
+
+	ctx := context.Background()
+	instanceDID := "did:web:coves.social"
+
+	// Setup identity resolver with local PLC
+	plcURL := os.Getenv("PLC_DIRECTORY_URL")
+	if plcURL == "" {
+		plcURL = "http://localhost:3002"
+	}
+	identityConfig := identity.DefaultConfig()
+	identityConfig.PLCURL = plcURL
+	identityResolver := identity.NewResolver(db, identityConfig)
+
+	// Setup services
+	communityRepo := postgres.NewCommunityRepository(db)
+	provisioner := communities.NewPDSAccountProvisioner("coves.social", pdsURL)
+	blobService := blobs.NewBlobService(pdsURL)
+
+	communityService := communities.NewCommunityServiceWithPDSFactory(
+		communityRepo,
+		pdsURL,
+		instanceDID,
+		"coves.social",
+		provisioner,
+		nil,
+		blobService,
+	)
+
+	consumer := jetstream.NewCommunityEventConsumer(communityRepo, instanceDID, true, identityResolver)
+
+	// Helper to wait for Jetstream update event and process it
+	waitForUpdateEvent := func(t *testing.T, communityDID string, timeout time.Duration) *jetstream.JetstreamEvent {
+		eventChan := make(chan *jetstream.JetstreamEvent, 10)
+		done := make(chan bool)
+		subscribeCtx, cancelSubscribe := context.WithTimeout(ctx, timeout)
+		defer cancelSubscribe()
+
+		go func() {
+			conn, _, dialErr := websocket.DefaultDialer.Dial(jetstreamURL, nil)
+			if dialErr != nil {
+				t.Logf("Failed to connect to Jetstream: %v", dialErr)
+				return
+			}
+			defer func() { _ = conn.Close() }()
+
+			for {
+				select {
+				case <-done:
+					return
+				case <-subscribeCtx.Done():
+					return
+				default:
+					if deadlineErr := conn.SetReadDeadline(time.Now().Add(5 * time.Second)); deadlineErr != nil {
+						return
+					}
+
+					var event jetstream.JetstreamEvent
+					if readErr := conn.ReadJSON(&event); readErr != nil {
+						continue
+					}
+
+					if event.Kind == "commit" && event.Commit != nil &&
+						event.Commit.Collection == "social.coves.community.profile" &&
+						event.Commit.Operation == "update" &&
+						event.Did == communityDID {
+						eventChan <- &event
+					}
+				}
+			}
+		}()
+
+		select {
+		case event := <-eventChan:
+			close(done)
+			return event
+		case <-time.After(timeout):
+			close(done)
+			return nil
+		}
+	}
+
+	t.Run("add banner to community without one", func(t *testing.T) {
+		uniqueName := fmt.Sprintf("ban%d", time.Now().UnixNano()%1000000)
+		creatorDID := "did:plc:banner-add-test"
+
+		// Create a community WITHOUT a banner
+		t.Logf("\n📝 Creating community without banner...")
+		community, createErr := communityService.CreateCommunity(ctx, communities.CreateCommunityRequest{
+			Name:                   uniqueName,
+			DisplayName:            "Community Without Banner",
+			Description:            "Will add banner via update",
+			Visibility:             "public",
+			CreatedByDID:           creatorDID,
+			HostedByDID:            instanceDID,
+			AllowExternalDiscovery: true,
+		})
+		if createErr != nil {
+			t.Fatalf("Failed to create community: %v", createErr)
+		}
+		t.Logf("✅ Community created: DID=%s", community.DID)
+
+		// Verify no banner initially
+		initial, err := communityService.GetCommunity(ctx, community.DID)
+		if err != nil {
+			t.Fatalf("Community not indexed: %v", err)
+		}
+		if initial.BannerCID != "" {
+			t.Fatalf("Expected no initial banner, got: %s", initial.BannerCID)
+		}
+		t.Logf("   Initial BannerCID: '' (confirmed empty)")
+
+		// Create test banner image (300x100 green rectangle)
+		bannerData := createTestPNGImage(300, 100, color.RGBA{0, 255, 0, 255})
+		t.Logf("\n📝 Updating community with banner (%d bytes)...", len(bannerData))
+
+		// Start listening for Jetstream event
+		eventReceived := make(chan *jetstream.JetstreamEvent, 1)
+		go func() {
+			event := waitForUpdateEvent(t, community.DID, 15*time.Second)
+			eventReceived <- event
+		}()
+		time.Sleep(500 * time.Millisecond) // Give subscriber time to connect
+
+		// Perform the update with banner
+		newDisplayName := "Community With New Banner"
+		updated, updateErr := communityService.UpdateCommunity(ctx, communities.UpdateCommunityRequest{
+			CommunityDID:   community.DID,
+			UpdatedByDID:   creatorDID,
+			DisplayName:    &newDisplayName,
+			BannerBlob:     bannerData,
+			BannerMimeType: "image/png",
+		})
+		if updateErr != nil {
+			t.Fatalf("Failed to update community with banner: %v", updateErr)
+		}
+
+		t.Logf("✅ Community update written to PDS:")
+		t.Logf("   New RecordCID: %s", updated.RecordCID)
+
+		// Wait for REAL Jetstream event
+		t.Logf("\n⏳ Waiting for update event from Jetstream...")
+		realEvent := <-eventReceived
+		if realEvent == nil {
+			t.Fatalf("Timeout waiting for Jetstream update event")
+		}
+
+		t.Logf("✅ Received REAL update event from Jetstream!")
+		t.Logf("   Operation: %s", realEvent.Commit.Operation)
+		t.Logf("   CID: %s", realEvent.Commit.CID)
+
+		// Extract banner CID from real event
+		var bannerCIDFromEvent string
+		if realEvent.Commit.Record != nil {
+			if banner, hasBanner := realEvent.Commit.Record["banner"].(map[string]interface{}); hasBanner {
+				t.Logf("   Banner in event: %v", banner)
+				if ref, hasRef := banner["ref"].(map[string]interface{}); hasRef {
+					if link, hasLink := ref["$link"].(string); hasLink {
+						bannerCIDFromEvent = link
+						t.Logf("   BannerCID from Jetstream: %s", bannerCIDFromEvent)
+					}
+				}
+			}
+		}
+
+		// Process the REAL event through consumer
+		t.Logf("\n🔄 Processing real Jetstream event through consumer...")
+		if handleErr := consumer.HandleEvent(ctx, realEvent); handleErr != nil {
+			t.Logf("   Consumer error: %v", handleErr)
+		}
+
+		// Verify banner CID is now set in DB
+		final, err := communityRepo.GetByDID(ctx, community.DID)
+		if err != nil {
+			t.Fatalf("Failed to get final community: %v", err)
+		}
+
+		t.Logf("\n✅ Community banner update verified:")
+		t.Logf("   DisplayName: %s", final.DisplayName)
+		t.Logf("   BannerCID in DB: %s", final.BannerCID)
+
+		if final.BannerCID == "" {
+			t.Errorf("Expected BannerCID to be set after update")
+		}
+
+		// Verify DB matches Jetstream event
+		if bannerCIDFromEvent != "" && final.BannerCID != bannerCIDFromEvent {
+			t.Errorf("BannerCID mismatch: DB has %s, Jetstream has %s", final.BannerCID, bannerCIDFromEvent)
+		} else if bannerCIDFromEvent != "" {
+			t.Logf("   ✅ BannerCID matches between DB and Jetstream!")
+		}
+
+		t.Logf("\n✅ TRUE E2E ADD BANNER FLOW COMPLETE")
+	})
+
+	t.Run("replace existing banner with new one", func(t *testing.T) {
+		uniqueName := fmt.Sprintf("rpban%d", time.Now().UnixNano()%1000000)
+		creatorDID := "did:plc:banner-replace-test"
+
+		// Create a community WITH an initial banner (red rectangle)
+		initialBannerData := createTestPNGImage(300, 100, color.RGBA{255, 0, 0, 255})
+		t.Logf("\n📝 Creating community with initial banner (red, %d bytes)...", len(initialBannerData))
+
+		community, createErr := communityService.CreateCommunity(ctx, communities.CreateCommunityRequest{
+			Name:                   uniqueName,
+			DisplayName:            "Community With Initial Banner",
+			Description:            "Will replace banner",
+			Visibility:             "public",
+			CreatedByDID:           creatorDID,
+			HostedByDID:            instanceDID,
+			AllowExternalDiscovery: true,
+			BannerBlob:             initialBannerData,
+			BannerMimeType:         "image/png",
+		})
+		if createErr != nil {
+			t.Fatalf("Failed to create community with banner: %v", createErr)
+		}
+		t.Logf("✅ Community created: DID=%s", community.DID)
+
+		// Verify initial banner is set
+		initial, err := communityService.GetCommunity(ctx, community.DID)
+		if err != nil {
+			t.Fatalf("Community not indexed: %v", err)
+		}
+		initialBannerCID := initial.BannerCID
+		if initialBannerCID == "" {
+			t.Fatalf("Expected initial banner to be set")
+		}
+		t.Logf("   Initial BannerCID: %s", initialBannerCID)
+
+		// Create NEW banner image (300x100 blue rectangle - different from initial red)
+		newBannerData := createTestPNGImage(300, 100, color.RGBA{0, 0, 255, 255})
+		t.Logf("\n📝 Replacing banner with new one (blue, %d bytes)...", len(newBannerData))
+
+		// Start listening for Jetstream event
+		eventReceived := make(chan *jetstream.JetstreamEvent, 1)
+		go func() {
+			event := waitForUpdateEvent(t, community.DID, 15*time.Second)
+			eventReceived <- event
+		}()
+		time.Sleep(500 * time.Millisecond)
+
+		// Perform the update with NEW banner
+		newDisplayName := "Community With Replaced Banner"
+		updated, updateErr := communityService.UpdateCommunity(ctx, communities.UpdateCommunityRequest{
+			CommunityDID:   community.DID,
+			UpdatedByDID:   creatorDID,
+			DisplayName:    &newDisplayName,
+			BannerBlob:     newBannerData,
+			BannerMimeType: "image/png",
+		})
+		if updateErr != nil {
+			t.Fatalf("Failed to update community with new banner: %v", updateErr)
+		}
+
+		t.Logf("✅ Community update written to PDS:")
+		t.Logf("   New RecordCID: %s", updated.RecordCID)
+
+		// Wait for REAL Jetstream event
+		t.Logf("\n⏳ Waiting for update event from Jetstream...")
+		realEvent := <-eventReceived
+		if realEvent == nil {
+			t.Fatalf("Timeout waiting for Jetstream update event")
+		}
+
+		t.Logf("✅ Received REAL update event from Jetstream!")
+		t.Logf("   Operation: %s", realEvent.Commit.Operation)
+
+		// Extract new banner CID from real event
+		var newBannerCIDFromEvent string
+		if realEvent.Commit.Record != nil {
+			if banner, hasBanner := realEvent.Commit.Record["banner"].(map[string]interface{}); hasBanner {
+				if ref, hasRef := banner["ref"].(map[string]interface{}); hasRef {
+					if link, hasLink := ref["$link"].(string); hasLink {
+						newBannerCIDFromEvent = link
+						t.Logf("   New BannerCID from Jetstream: %s", newBannerCIDFromEvent)
+					}
+				}
+			}
+		}
+
+		// Process the REAL event through consumer
+		t.Logf("\n🔄 Processing real Jetstream event through consumer...")
+		if handleErr := consumer.HandleEvent(ctx, realEvent); handleErr != nil {
+			t.Logf("   Consumer error: %v", handleErr)
+		}
+
+		// Verify banner CID has CHANGED
+		final, err := communityRepo.GetByDID(ctx, community.DID)
+		if err != nil {
+			t.Fatalf("Failed to get final community: %v", err)
+		}
+
+		t.Logf("\n✅ Community banner replacement verified:")
+		t.Logf("   DisplayName: %s", final.DisplayName)
+		t.Logf("   Old BannerCID: %s", initialBannerCID)
+		t.Logf("   New BannerCID: %s", final.BannerCID)
+
+		if final.BannerCID == "" {
+			t.Errorf("Expected BannerCID to be set after replacement")
+		}
+
+		if final.BannerCID == initialBannerCID {
+			t.Errorf("BannerCID should have changed after replacement! Old: %s, New: %s", initialBannerCID, final.BannerCID)
+		} else {
+			t.Logf("   ✅ BannerCID successfully changed!")
+		}
+
+		// Verify DB matches Jetstream event
+		if newBannerCIDFromEvent != "" && final.BannerCID != newBannerCIDFromEvent {
+			t.Errorf("BannerCID mismatch: DB has %s, Jetstream has %s", final.BannerCID, newBannerCIDFromEvent)
+		} else if newBannerCIDFromEvent != "" {
+			t.Logf("   ✅ New BannerCID matches between DB and Jetstream!")
+		}
+
+		t.Logf("\n✅ TRUE E2E REPLACE BANNER FLOW COMPLETE")
+	})
+}

tests/integration/community_e2e_test.go

···
 	provisioner := communities.NewPDSAccountProvisioner(instanceDomain, pdsURL)
 
 	// Create service with PDS factory for password-based auth in tests
-	communityService := communities.NewCommunityServiceWithPDSFactory(communityRepo, pdsURL, instanceDID, instanceDomain, provisioner, CommunityPasswordAuthPDSClientFactory())
+	communityService := communities.NewCommunityServiceWithPDSFactory(communityRepo, pdsURL, instanceDID, instanceDomain, provisioner, CommunityPasswordAuthPDSClientFactory(), nil)
 	if svc, ok := communityService.(interface{ SetPDSAccessToken(string) }); ok {
 		svc.SetPDSAccessToken(accessToken)
 	}

tests/integration/community_identifier_resolution_test.go

···
 		instanceDomain,
 		provisioner,
 		nil,
+		nil,
 	)
 
 	// Create a test community to resolve
···
 		instanceDID,
 		instanceDomain,
 		provisioner,
+		nil,
 		nil,
 	)
 
···
 		instanceDomain,
 		provisioner,
 		nil,
+		nil,
 	)
 
 	t.Run("DID error includes identifier", func(t *testing.T) {
···
 		instanceDID,
 		instanceDomain,
 		provisioner,
+		nil,
 		nil,
 	)
 

tests/integration/community_provisioning_test.go

···
 		"test.local",            // instanceDomain
 		provisioner,
 		nil,
+		nil,
 	)
 	ctx := context.Background()
 

tests/integration/community_service_integration_test.go

···
 			"did:web:coves.social",
 			"coves.social",
 			provisioner,
-		nil,
+			nil,
+			nil,
 		)
 
 		// Generate unique community name (keep short for DNS label limit)
···
 			"did:web:coves.social",
 			"coves.social",
 			provisioner,
-		nil,
+			nil,
+			nil,
 		)
 
 		// Try to create community with invalid name (should fail validation before PDS)
···
 			"did:web:coves.social",
 			"coves.social",
 			provisioner,
-		nil,
+			nil,
+			nil,
 		)
 
 		// Try 64-char name (exceeds DNS limit of 63)
···
 		"did:web:coves.social",
 		"coves.social",
 		provisioner,
+		nil,
 		nil,
 	)
 
···
 		"did:web:coves.social",
 		"coves.social",
 		provisioner,
+		nil,
 		nil,
 	)
 

tests/integration/community_update_e2e_test.go

···
 import (
 	"Coves/internal/atproto/identity"
 	"Coves/internal/atproto/jetstream"
+	"Coves/internal/core/blobs"
 	"Coves/internal/core/communities"
 	"Coves/internal/db/postgres"
 	"context"
···
 	// Setup services
 	communityRepo := postgres.NewCommunityRepository(db)
 	provisioner := communities.NewPDSAccountProvisioner("coves.social", pdsURL)
+	blobService := blobs.NewBlobService(pdsURL)
 	communityService := communities.NewCommunityServiceWithPDSFactory(
 		communityRepo,
 		pdsURL,
···
 		"coves.social",
 		provisioner,
 		nil,
+		blobService,
 	)
 
 	consumer := jetstream.NewCommunityEventConsumer(communityRepo, instanceDID, true, identityResolver)

tests/integration/feed_test.go

···
 		"test.coves.social",
 		nil,
 		nil,
+		nil,
 	)
 	feedService := communityFeeds.NewCommunityFeedService(feedRepo, communityService)
 	handler := communityFeed.NewGetCommunityHandler(feedService, nil, nil)
···
 		"test.coves.social",
 		nil,
 		nil,
+		nil,
 	)
 	feedService := communityFeeds.NewCommunityFeedService(feedRepo, communityService)
 	handler := communityFeed.NewGetCommunityHandler(feedService, nil, nil)
···
 		"http://localhost:3001",
 		"did:web:test.coves.social",
 		"test.coves.social",
+		nil,
 		nil,
 		nil,
 	)
···
 		"test.coves.social",
 		nil,
 		nil,
+		nil,
 	)
 	feedService := communityFeeds.NewCommunityFeedService(feedRepo, communityService)
 	handler := communityFeed.NewGetCommunityHandler(feedService, nil, nil)
···
 		"test.coves.social",
 		nil,
 		nil,
+		nil,
 	)
 	feedService := communityFeeds.NewCommunityFeedService(feedRepo, communityService)
 	handler := communityFeed.NewGetCommunityHandler(feedService, nil, nil)
···
 		"http://localhost:3001",
 		"did:web:test.coves.social",
 		"test.coves.social",
+		nil,
 		nil,
 		nil,
 	)
···
 		"test.coves.social",
 		nil,
 		nil,
+		nil,
 	)
 	feedService := communityFeeds.NewCommunityFeedService(feedRepo, communityService)
 	handler := communityFeed.NewGetCommunityHandler(feedService, nil, nil)
···
 		"http://localhost:3001",
 		"did:web:test.coves.social",
 		"test.coves.social",
+		nil,
 		nil,
 		nil,
 	)
···
 		"http://localhost:3001",
 		"did:web:test.coves.social",
 		"test.coves.social",
+		nil,
 		nil,
 		nil,
 	)
···
 		"test.coves.social",
 		nil,
 		nil,
+		nil,
 	)
 	feedService := communityFeeds.NewCommunityFeedService(feedRepo, communityService)
 	handler := communityFeed.NewGetCommunityHandler(feedService, nil, nil)
···
 		"test.coves.social",
 		nil,
 		nil,
+		nil,
 	)
 	feedService := communityFeeds.NewCommunityFeedService(feedRepo, communityService)
 	handler := communityFeed.NewGetCommunityHandler(feedService, nil, nil)
···
 		"http://localhost:3001",
 		"did:web:test.coves.social",
 		"test.coves.social",
+		nil,
 		nil,
 		nil,
 	)

tests/integration/post_creation_test.go

···
 		"test.coves.social",
 		nil, // provisioner
 		nil, // pdsClientFactory
+		nil, // blobService
 	)
 
 	postRepo := postgres.NewPostRepository(db)

tests/integration/post_e2e_test.go

···
 		pdsURL,
 		instanceDID,
 		instanceDomain,
-		provisioner, // ✅ Real provisioner for creating communities on PDS
+		provisioner, // Real provisioner for creating communities on PDS
 		nil,         // No PDS factory needed - no subscribe/block in this test
+		nil,         // No blob service for this test
 	)
 
 	postService := posts.NewPostService(postRepo, communityService, nil, nil, nil, nil, pdsURL) // nil aggregatorService, blobService, unfurlService, blueskyService for user-only tests

tests/integration/post_handler_test.go

···
 		"test.coves.social",
 		nil,
 		nil,
+		nil,
 	)
 
 	postRepo := postgres.NewPostRepository(db)
···
 		"test.coves.social",
 		nil,
 		nil,
+		nil,
 	)
 
 	postRepo := postgres.NewPostRepository(db)
···
 		"http://localhost:3001",
 		"did:web:test.coves.social",
 		"test.coves.social",
+		nil,
 		nil,
 		nil,
 	)

tests/integration/post_thumb_validation_test.go

···
 		"test.coves.social",
 		nil,
 		nil,
+		nil,
 	)
 
 	postRepo := postgres.NewPostRepository(db)

tests/integration/post_unfurl_test.go

···
 		"test.coves.social",
 		nil,
 		nil,
+		nil,
 	)
 
 	postService := posts.NewPostService(
···
 		"test.coves.social",
 		nil,
 		nil,
+		nil,
 	)
 
 	// Create post service WITHOUT unfurl service
···
 		"http://localhost:3001",
 		"did:web:test.coves.social",
 		"test.coves.social",
+		nil,
 		nil,
 		nil,
 	)
···
 		"http://localhost:3001",
 		"did:web:test.coves.social",
 		"test.coves.social",
+		nil,
 		nil,
 		nil,
 	)

tests/integration/user_journey_e2e_test.go

···
 	}
 
 	provisioner := communities.NewPDSAccountProvisioner(instanceDomain, pdsURL)
-	communityService := communities.NewCommunityServiceWithPDSFactory(communityRepo, pdsURL, instanceDID, instanceDomain, provisioner, CommunityPasswordAuthPDSClientFactory())
+	communityService := communities.NewCommunityServiceWithPDSFactory(communityRepo, pdsURL, instanceDID, instanceDomain, provisioner, CommunityPasswordAuthPDSClientFactory(), nil)
 	postService := posts.NewPostService(postRepo, communityService, nil, nil, nil, nil, pdsURL)
 	timelineService := timelineCore.NewTimelineService(timelineRepo)