commit 0dad0018e29a1b8650342cca187800e86c01a511 · bretton.dev/coves

+41 -23

cmd/server/main.go

···

       1
        
       package main

     

       2
        
       

     

       3
        
       import (

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       4
        
       	"bytes"

     

       5
        
       	"context"

     

       6
        
       	"database/sql"

     
···

       18
        
       	_ "github.com/lib/pq"

     

       19
        
       	"github.com/pressly/goose/v3"

     

       20
        
       

     

       21
       -
       	"Coves/internal/api/handlers/oauth"

     

       22
       -
       	"Coves/internal/api/middleware"

     

       23
       -
       	"Coves/internal/api/routes"

     

       24
       -
       	"Coves/internal/atproto/did"

     

       25
       -
       	"Coves/internal/atproto/identity"

     

       26
       -
       	"Coves/internal/atproto/jetstream"

     

       27
       -
       	"Coves/internal/core/communities"

     

       28
        
       	oauthCore "Coves/internal/core/oauth"

     

       29
       -
       	"Coves/internal/core/users"

     

       30
        
       	postgresRepo "Coves/internal/db/postgres"

     

       31
        
       )

     

       32
        
       

     
···

       48
        
       	if err != nil {

     

       49
        
       		log.Fatal("Failed to connect to database:", err)

     

       50
        
       	}

     

       51
       -
       	defer db.Close()

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       52
        
       

     

       53
       -
       	if err := db.Ping(); err != nil {

     

       54
        
       		log.Fatal("Failed to ping database:", err)

     

       55
        
       	}

     

       56
        
       

     

       57
        
       	log.Println("Connected to AppView database")

     

       58
        
       

     

       59
        
       	// Run migrations

     

       60
       -
       	if err := goose.SetDialect("postgres"); err != nil {

     

       61
        
       		log.Fatal("Failed to set goose dialect:", err)

     

       62
        
       	}

     

       63
        
       

     

       64
       -
       	if err := goose.Up(db, "internal/db/migrations"); err != nil {

     

       65
        
       		log.Fatal("Failed to run migrations:", err)

     

       66
        
       	}

     

       67
        
       

     
···

       84
        
       		identityConfig.PLCURL = plcURL

     

       85
        
       	}

     

       86
        
       	if cacheTTL := os.Getenv("IDENTITY_CACHE_TTL"); cacheTTL != "" {

     

       87
       -
       		if duration, err := time.ParseDuration(cacheTTL); err == nil {

     

       88
        
       			identityConfig.CacheTTL = duration

     

       89
        
       		}

     

       90
        
       	}

     
···

       154
        
       	pdsPassword := os.Getenv("PDS_INSTANCE_PASSWORD")

     

       155
        
       	if pdsHandle != "" && pdsPassword != "" {

     

       156
        
       		log.Printf("Authenticating Coves instance (%s) with PDS...", instanceDID)

     

       157
       -
       		accessToken, err := authenticateWithPDS(defaultPDS, pdsHandle, pdsPassword)

     

       158
       -
       		if err != nil {

     

       159
       -
       			log.Printf("Warning: Failed to authenticate with PDS: %v", err)

     

       160
        
       			log.Println("Community creation will fail until PDS authentication is configured")

     

       161
        
       		} else {

     

       162
        
       			if svc, ok := communityService.(interface{ SetPDSAccessToken(string) }); ok {

     
···

       180
        
       	userConsumer := jetstream.NewUserEventConsumer(userService, identityResolver, jetstreamURL, pdsFilter)

     

       181
        
       	ctx := context.Background()

     

       182
        
       	go func() {

     

       183
       -
       		if err := userConsumer.Start(ctx); err != nil {

     

       184
       -
       			log.Printf("Jetstream consumer stopped: %v", err)

     

       185
        
       		}

     

       186
        
       	}()

     

       187
        
       

     
···

       199
        
       		defer ticker.Stop()

     

       200
        
       		for range ticker.C {

     

       201
        
       			if pgStore, ok := sessionStore.(*oauthCore.PostgresSessionStore); ok {

     

       202
       -
       				_ = pgStore.CleanupExpiredRequests(ctx)

     

       203
       -
       				_ = pgStore.CleanupExpiredSessions(ctx)

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       204
        
       				log.Println("OAuth cleanup completed")

     

       205
        
       			}

     

       206
        
       		}

     
···

       242
        
       

     

       243
        
       	r.Get("/health", func(w http.ResponseWriter, r *http.Request) {

     

       244
        
       		w.WriteHeader(http.StatusOK)

     

       245
       -
       		w.Write([]byte("OK"))

     

       0
        
       
     

       0
        
       
     

       246
        
       	})

     

       247
        
       

     

       248
        
       	port := os.Getenv("APPVIEW_PORT")

     
···

       284
        
       	if err != nil {

     

       285
        
       		return "", fmt.Errorf("failed to call PDS: %w", err)

     

       286
        
       	}

     

       287
       -
       	defer resp.Body.Close()

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       288
        
       

     

       289
        
       	if resp.StatusCode != http.StatusOK {

     

       290
       -
       		body, _ := io.ReadAll(resp.Body)

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       291
        
       		return "", fmt.Errorf("PDS returned status %d: %s", resp.StatusCode, string(body))

     

       292
        
       	}

     

       293

···

       1
        
       package main

     

       2
        
       

     

       3
        
       import (

     

       4
       +
       	"Coves/internal/api/handlers/oauth"

     

       5
       +
       	"Coves/internal/api/middleware"

     

       6
       +
       	"Coves/internal/api/routes"

     

       7
       +
       	"Coves/internal/atproto/did"

     

       8
       +
       	"Coves/internal/atproto/identity"

     

       9
       +
       	"Coves/internal/atproto/jetstream"

     

       10
       +
       	"Coves/internal/core/communities"

     

       11
       +
       	"Coves/internal/core/users"

     

       12
        
       	"bytes"

     

       13
        
       	"context"

     

       14
        
       	"database/sql"

     
···

       26
        
       	_ "github.com/lib/pq"

     

       27
        
       	"github.com/pressly/goose/v3"

     

       28
        
       

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       29
        
       	oauthCore "Coves/internal/core/oauth"

     

       30
       +
       

     

       31
        
       	postgresRepo "Coves/internal/db/postgres"

     

       32
        
       )

     

       33
        
       

     
···

       49
        
       	if err != nil {

     

       50
        
       		log.Fatal("Failed to connect to database:", err)

     

       51
        
       	}

     

       52
       +
       	defer func() {

     

       53
       +
       		if closeErr := db.Close(); closeErr != nil {

     

       54
       +
       			log.Printf("Failed to close database connection: %v", closeErr)

     

       55
       +
       		}

     

       56
       +
       	}()

     

       57
        
       

     

       58
       +
       	if err = db.Ping(); err != nil {

     

       59
        
       		log.Fatal("Failed to ping database:", err)

     

       60
        
       	}

     

       61
        
       

     

       62
        
       	log.Println("Connected to AppView database")

     

       63
        
       

     

       64
        
       	// Run migrations

     

       65
       +
       	if err = goose.SetDialect("postgres"); err != nil {

     

       66
        
       		log.Fatal("Failed to set goose dialect:", err)

     

       67
        
       	}

     

       68
        
       

     

       69
       +
       	if err = goose.Up(db, "internal/db/migrations"); err != nil {

     

       70
        
       		log.Fatal("Failed to run migrations:", err)

     

       71
        
       	}

     

       72
        
       

     
···

       89
        
       		identityConfig.PLCURL = plcURL

     

       90
        
       	}

     

       91
        
       	if cacheTTL := os.Getenv("IDENTITY_CACHE_TTL"); cacheTTL != "" {

     

       92
       +
       		if duration, parseErr := time.ParseDuration(cacheTTL); parseErr == nil {

     

       93
        
       			identityConfig.CacheTTL = duration

     

       94
        
       		}

     

       95
        
       	}

     
···

       159
        
       	pdsPassword := os.Getenv("PDS_INSTANCE_PASSWORD")

     

       160
        
       	if pdsHandle != "" && pdsPassword != "" {

     

       161
        
       		log.Printf("Authenticating Coves instance (%s) with PDS...", instanceDID)

     

       162
       +
       		accessToken, authErr := authenticateWithPDS(defaultPDS, pdsHandle, pdsPassword)

     

       163
       +
       		if authErr != nil {

     

       164
       +
       			log.Printf("Warning: Failed to authenticate with PDS: %v", authErr)

     

       165
        
       			log.Println("Community creation will fail until PDS authentication is configured")

     

       166
        
       		} else {

     

       167
        
       			if svc, ok := communityService.(interface{ SetPDSAccessToken(string) }); ok {

     
···

       185
        
       	userConsumer := jetstream.NewUserEventConsumer(userService, identityResolver, jetstreamURL, pdsFilter)

     

       186
        
       	ctx := context.Background()

     

       187
        
       	go func() {

     

       188
       +
       		if startErr := userConsumer.Start(ctx); startErr != nil {

     

       189
       +
       			log.Printf("Jetstream consumer stopped: %v", startErr)

     

       190
        
       		}

     

       191
        
       	}()

     

       192
        
       

     
···

       204
        
       		defer ticker.Stop()

     

       205
        
       		for range ticker.C {

     

       206
        
       			if pgStore, ok := sessionStore.(*oauthCore.PostgresSessionStore); ok {

     

       207
       +
       				if cleanupErr := pgStore.CleanupExpiredRequests(ctx); cleanupErr != nil {

     

       208
       +
       					log.Printf("Failed to cleanup expired OAuth requests: %v", cleanupErr)

     

       209
       +
       				}

     

       210
       +
       				if cleanupErr := pgStore.CleanupExpiredSessions(ctx); cleanupErr != nil {

     

       211
       +
       					log.Printf("Failed to cleanup expired OAuth sessions: %v", cleanupErr)

     

       212
       +
       				}

     

       213
        
       				log.Println("OAuth cleanup completed")

     

       214
        
       			}

     

       215
        
       		}

     
···

       251
        
       

     

       252
        
       	r.Get("/health", func(w http.ResponseWriter, r *http.Request) {

     

       253
        
       		w.WriteHeader(http.StatusOK)

     

       254
       +
       		if _, err := w.Write([]byte("OK")); err != nil {

     

       255
       +
       			log.Printf("Failed to write health check response: %v", err)

     

       256
       +
       		}

     

       257
        
       	})

     

       258
        
       

     

       259
        
       	port := os.Getenv("APPVIEW_PORT")

     
···

       295
        
       	if err != nil {

     

       296
        
       		return "", fmt.Errorf("failed to call PDS: %w", err)

     

       297
        
       	}

     

       298
       +
       	defer func() {

     

       299
       +
       		if closeErr := resp.Body.Close(); closeErr != nil {

     

       300
       +
       			log.Printf("Failed to close response body: %v", closeErr)

     

       301
       +
       		}

     

       302
       +
       	}()

     

       303
        
       

     

       304
        
       	if resp.StatusCode != http.StatusOK {

     

       305
       +
       		body, readErr := io.ReadAll(resp.Body)

     

       306
       +
       		if readErr != nil {

     

       307
       +
       			return "", fmt.Errorf("PDS returned status %d and failed to read body: %w", resp.StatusCode, readErr)

     

       308
       +
       		}

     

       309
        
       		return "", fmt.Errorf("PDS returned status %d: %s", resp.StatusCode, string(body))

     

       310
        
       	}

     

       311

+62 -53

cmd/validate-lexicon/main.go

···

       88
        
       		// Only process .json files

     

       89
        
       		if !info.IsDir() && filepath.Ext(path) == ".json" {

     

       90
        
       			schemaFiles = append(schemaFiles, path)

     

       91
       -
       			

     

       92
        
       			// Convert file path to schema ID

     

       93
        
       			// e.g., internal/atproto/lexicon/social/coves/actor/profile.json -> social.coves.actor.profile

     

       94
       -
       			relPath, _ := filepath.Rel(schemaPath, path)

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       95
        
       			schemaID := filepath.ToSlash(relPath)

     

       96
        
       			schemaID = schemaID[:len(schemaID)-5] // Remove .json extension

     

       97
        
       			schemaID = strings.ReplaceAll(schemaID, "/", ".")

     
···

       99
        
       		}

     

       100
        
       		return nil

     

       101
        
       	})

     

       102
       -
       

     

       103
        
       	if err != nil {

     

       104
        
       		return fmt.Errorf("error walking schema directory: %w", err)

     

       105
        
       	}

     
···

       157
        
       		}

     

       158
        
       		return nil

     

       159
        
       	})

     

       160
       -
       

     

       161
        
       	if err != nil {

     

       162
        
       		return fmt.Errorf("error walking schema directory: %w", err)

     

       163
        
       	}

     
···

       182
        
       // extractAllSchemaIDs walks the schema directory and returns all schema IDs

     

       183
        
       func extractAllSchemaIDs(schemaPath string) []string {

     

       184
        
       	var schemaIDs []string

     

       185
       -
       	

     

       186
       -
       	filepath.Walk(schemaPath, func(path string, info os.FileInfo, err error) error {

     

       187
        
       		if err != nil {

     

       188
        
       			return err

     

       189
        
       		}

     

       190
       -
       		

     

       191
        
       		// Skip test-data directory

     

       192
        
       		if info.IsDir() && info.Name() == "test-data" {

     

       193
        
       			return filepath.SkipDir

     

       194
        
       		}

     

       195
       -
       		

     

       196
        
       		// Only process .json files

     

       197
        
       		if !info.IsDir() && filepath.Ext(path) == ".json" {

     

       198
        
       			// Convert file path to schema ID

     

       199
       -
       			relPath, _ := filepath.Rel(schemaPath, path)

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       200
        
       			schemaID := filepath.ToSlash(relPath)

     

       201
        
       			schemaID = schemaID[:len(schemaID)-5] // Remove .json extension

     

       202
        
       			schemaID = strings.ReplaceAll(schemaID, "/", ".")

     

       203
       -
       			

     

       204
        
       			// Only include record schemas (not procedures)

     

       205
       -
       			if strings.Contains(schemaID, ".record") || 

     

       206
       -
       			   strings.Contains(schemaID, ".profile") || 

     

       207
       -
       			   strings.Contains(schemaID, ".rules") || 

     

       208
       -
       			   strings.Contains(schemaID, ".wiki") || 

     

       209
       -
       			   strings.Contains(schemaID, ".subscription") || 

     

       210
       -
       			   strings.Contains(schemaID, ".membership") ||

     

       211
       -
       			   strings.Contains(schemaID, ".vote") ||

     

       212
       -
       			   strings.Contains(schemaID, ".tag") ||

     

       213
       -
       			   strings.Contains(schemaID, ".comment") ||

     

       214
       -
       			   strings.Contains(schemaID, ".share") ||

     

       215
       -
       			   strings.Contains(schemaID, ".tribunalVote") ||

     

       216
       -
       			   strings.Contains(schemaID, ".ruleProposal") ||

     

       217
       -
       			   strings.Contains(schemaID, ".ban") {

     

       218
        
       				schemaIDs = append(schemaIDs, schemaID)

     

       219
        
       			}

     

       220
        
       		}

     

       221
        
       		return nil

     

       222
       -
       	})

     

       223
       -
       	

     

       0
        
       
     

       0
        
       
     

       224
        
       	return schemaIDs

     

       225
        
       }

     

       226
        
       

     

       227
        
       // validateTestData validates test JSON data files against their corresponding schemas

     

       228
       -
       func validateTestData(catalog *lexicon.BaseCatalog, testDataPath string, verbose bool, strict bool, allSchemas []string) error {

     

       229
        
       	// Check if test data directory exists

     

       230
        
       	if _, err := os.Stat(testDataPath); os.IsNotExist(err) {

     

       231
        
       		return fmt.Errorf("test data path does not exist: %s", testDataPath)

     
···

       248
        
       		if !info.IsDir() && filepath.Ext(path) == ".json" {

     

       249
        
       			filename := filepath.Base(path)

     

       250
        
       			isInvalidTest := strings.Contains(filename, "-invalid-")

     

       251
       -
       			

     

       252
        
       			if verbose {

     

       253
        
       				if isInvalidTest {

     

       254
        
       					fmt.Printf("\n  Testing (expect failure): %s\n", filename)

     
···

       263
        
       				validationErrors = append(validationErrors, fmt.Sprintf("Failed to open %s: %v", path, err))

     

       264
        
       				return nil

     

       265
        
       			}

     

       266
       -
       			defer file.Close()

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       267
        
       

     

       268
       -
       			data, err := io.ReadAll(file)

     

       269
       -
       			if err != nil {

     

       270
       -
       				validationErrors = append(validationErrors, fmt.Sprintf("Failed to read %s: %v", path, err))

     

       271
        
       				return nil

     

       272
        
       			}

     

       273
        
       

     
···

       275
        
       			var recordData map[string]interface{}

     

       276
        
       			decoder := json.NewDecoder(bytes.NewReader(data))

     

       277
        
       			decoder.UseNumber() // This preserves numbers as json.Number instead of float64

     

       278
       -
       			if err := decoder.Decode(&recordData); err != nil {

     

       279
       -
       				validationErrors = append(validationErrors, fmt.Sprintf("Failed to parse JSON in %s: %v", path, err))

     

       280
        
       				return nil

     

       281
        
       			}

     

       282
       -
       			

     

       283
        
       			// Convert json.Number values to appropriate types

     

       284
        
       			recordData = convertNumbers(recordData).(map[string]interface{})

     

       285
        
       

     
···

       299
        
       			}

     

       300
        
       

     

       301
        
       			// Validate the record

     

       302
       -
       			err = lexicon.ValidateRecord(catalog, recordData, recordType, flags)

     

       303
       -
       			

     

       304
        
       			if isInvalidTest {

     

       305
        
       				// This file should fail validation

     

       306
        
       				invalidFiles++

     

       307
       -
       				if err != nil {

     

       308
        
       					invalidFailCount++

     

       309
        
       					if verbose {

     

       310
       -
       						fmt.Printf("    ✅ Correctly rejected invalid %s record: %v\n", recordType, err)

     

       311
        
       					}

     

       312
        
       				} else {

     

       313
        
       					validationErrors = append(validationErrors, fmt.Sprintf("Invalid test file %s passed validation when it should have failed", path))

     
···

       318
        
       			} else {

     

       319
        
       				// This file should pass validation

     

       320
        
       				validFiles++

     

       321
       -
       				if err != nil {

     

       322
       -
       					validationErrors = append(validationErrors, fmt.Sprintf("Validation failed for %s (type: %s): %v", path, recordType, err))

     

       323
        
       					if verbose {

     

       324
       -
       						fmt.Printf("    ❌ Failed: %v\n", err)

     

       325
        
       					}

     

       326
        
       				} else {

     

       327
        
       					validSuccessCount++

     
···

       334
        
       		}

     

       335
        
       		return nil

     

       336
        
       	})

     

       337
       -
       

     

       338
        
       	if err != nil {

     

       339
        
       		return fmt.Errorf("error walking test data directory: %w", err)

     

       340
        
       	}

     
···

       355
        
       		fmt.Printf("\n📋 Validation Summary:\n")

     

       356
        
       		fmt.Printf("  Valid test files:   %d/%d passed\n", validSuccessCount, validFiles)

     

       357
        
       		fmt.Printf("  Invalid test files: %d/%d correctly rejected\n", invalidFailCount, invalidFiles)

     

       358
       -
       		

     

       359
        
       		if validSuccessCount == validFiles && invalidFailCount == invalidFiles {

     

       360
        
       			fmt.Printf("\n  ✅ All test files behaved as expected!\n")

     

       361
        
       		}

     

       362
       -
       		

     

       363
        
       		// Show test coverage summary (only for valid files)

     

       364
        
       		fmt.Printf("\n📊 Test Data Coverage Summary:\n")

     

       365
        
       		fmt.Printf("  - Records with test data: %d types\n", len(testedTypes))

     

       366
        
       		fmt.Printf("  - Valid test files: %d\n", validFiles)

     

       367
        
       		fmt.Printf("  - Invalid test files: %d (for error validation)\n", invalidFiles)

     

       368
       -
       		

     

       369
        
       		fmt.Printf("\n  Tested record types:\n")

     

       370
        
       		for recordType := range testedTypes {

     

       371
        
       			fmt.Printf("    ✓ %s\n", recordType)

     

       372
        
       		}

     

       373
       -
       		

     

       374
        
       		// Show untested schemas

     

       375
        
       		untestedCount := 0

     

       376
        
       		fmt.Printf("\n  ⚠️  Record types without test data:\n")

     
···

       380
        
       				untestedCount++

     

       381
        
       			}

     

       382
        
       		}

     

       383
       -
       		

     

       384
        
       		if untestedCount == 0 {

     

       385
        
       			fmt.Println("    (None - full test coverage!)")

     

       386
        
       		} else {

     

       387
       -
       			fmt.Printf("\n  Coverage: %d/%d record types have test data (%.1f%%)\n", 

     

       388
       -
       				len(testedTypes), len(allSchemas), 

     

       389
        
       				float64(len(testedTypes))/float64(len(allSchemas))*100)

     

       390
        
       		}

     

       391
        
       	}

     
···

       404
        
       		"social.coves.richtext.facet#italic",

     

       405
        
       		"social.coves.richtext.facet#strikethrough",

     

       406
        
       		"social.coves.richtext.facet#spoiler",

     

       407
       -
       		

     

       408
        
       		// Post types and views

     

       409
        
       		"social.coves.post.get#postView",

     

       410
        
       		"social.coves.post.get#authorView",

     
···

       414
        
       		"social.coves.post.get#externalView",

     

       415
        
       		"social.coves.post.get#postStats",

     

       416
        
       		"social.coves.post.get#viewerState",

     

       417
       -
       		

     

       418
        
       		// Post record types

     

       419
        
       		"social.coves.post.record#originalAuthor",

     

       420
       -
       		

     

       421
        
       		// Actor definitions

     

       422
        
       		"social.coves.actor.profile#geoLocation",

     

       423
       -
       		

     

       424
        
       		// Community definitions

     

       425
        
       		"social.coves.community.rules#rule",

     

       426
        
       	}

···

       88
        
       		// Only process .json files

     

       89
        
       		if !info.IsDir() && filepath.Ext(path) == ".json" {

     

       90
        
       			schemaFiles = append(schemaFiles, path)

     

       91
       +
       

     

       92
        
       			// Convert file path to schema ID

     

       93
        
       			// e.g., internal/atproto/lexicon/social/coves/actor/profile.json -> social.coves.actor.profile

     

       94
       +
       			relPath, err := filepath.Rel(schemaPath, path)

     

       95
       +
       			if err != nil {

     

       96
       +
       				return fmt.Errorf("failed to compute relative path: %w", err)

     

       97
       +
       			}

     

       98
        
       			schemaID := filepath.ToSlash(relPath)

     

       99
        
       			schemaID = schemaID[:len(schemaID)-5] // Remove .json extension

     

       100
        
       			schemaID = strings.ReplaceAll(schemaID, "/", ".")

     
···

       102
        
       		}

     

       103
        
       		return nil

     

       104
        
       	})

     

       0
        
       
     

       105
        
       	if err != nil {

     

       106
        
       		return fmt.Errorf("error walking schema directory: %w", err)

     

       107
        
       	}

     
···

       159
        
       		}

     

       160
        
       		return nil

     

       161
        
       	})

     

       0
        
       
     

       162
        
       	if err != nil {

     

       163
        
       		return fmt.Errorf("error walking schema directory: %w", err)

     

       164
        
       	}

     
···

       183
        
       // extractAllSchemaIDs walks the schema directory and returns all schema IDs

     

       184
        
       func extractAllSchemaIDs(schemaPath string) []string {

     

       185
        
       	var schemaIDs []string

     

       186
       +
       

     

       187
       +
       	if err := filepath.Walk(schemaPath, func(path string, info os.FileInfo, err error) error {

     

       188
        
       		if err != nil {

     

       189
        
       			return err

     

       190
        
       		}

     

       191
       +
       

     

       192
        
       		// Skip test-data directory

     

       193
        
       		if info.IsDir() && info.Name() == "test-data" {

     

       194
        
       			return filepath.SkipDir

     

       195
        
       		}

     

       196
       +
       

     

       197
        
       		// Only process .json files

     

       198
        
       		if !info.IsDir() && filepath.Ext(path) == ".json" {

     

       199
        
       			// Convert file path to schema ID

     

       200
       +
       			relPath, err := filepath.Rel(schemaPath, path)

     

       201
       +
       			if err != nil {

     

       202
       +
       				return err

     

       203
       +
       			}

     

       204
        
       			schemaID := filepath.ToSlash(relPath)

     

       205
        
       			schemaID = schemaID[:len(schemaID)-5] // Remove .json extension

     

       206
        
       			schemaID = strings.ReplaceAll(schemaID, "/", ".")

     

       207
       +
       

     

       208
        
       			// Only include record schemas (not procedures)

     

       209
       +
       			if strings.Contains(schemaID, ".record") ||

     

       210
       +
       				strings.Contains(schemaID, ".profile") ||

     

       211
       +
       				strings.Contains(schemaID, ".rules") ||

     

       212
       +
       				strings.Contains(schemaID, ".wiki") ||

     

       213
       +
       				strings.Contains(schemaID, ".subscription") ||

     

       214
       +
       				strings.Contains(schemaID, ".membership") ||

     

       215
       +
       				strings.Contains(schemaID, ".vote") ||

     

       216
       +
       				strings.Contains(schemaID, ".tag") ||

     

       217
       +
       				strings.Contains(schemaID, ".comment") ||

     

       218
       +
       				strings.Contains(schemaID, ".share") ||

     

       219
       +
       				strings.Contains(schemaID, ".tribunalVote") ||

     

       220
       +
       				strings.Contains(schemaID, ".ruleProposal") ||

     

       221
       +
       				strings.Contains(schemaID, ".ban") {

     

       222
        
       				schemaIDs = append(schemaIDs, schemaID)

     

       223
        
       			}

     

       224
        
       		}

     

       225
        
       		return nil

     

       226
       +
       	}); err != nil {

     

       227
       +
       		log.Printf("Warning: failed to walk schema directory: %v", err)

     

       228
       +
       	}

     

       229
       +
       

     

       230
        
       	return schemaIDs

     

       231
        
       }

     

       232
        
       

     

       233
        
       // validateTestData validates test JSON data files against their corresponding schemas

     

       234
       +
       func validateTestData(catalog *lexicon.BaseCatalog, testDataPath string, verbose, strict bool, allSchemas []string) error {

     

       235
        
       	// Check if test data directory exists

     

       236
        
       	if _, err := os.Stat(testDataPath); os.IsNotExist(err) {

     

       237
        
       		return fmt.Errorf("test data path does not exist: %s", testDataPath)

     
···

       254
        
       		if !info.IsDir() && filepath.Ext(path) == ".json" {

     

       255
        
       			filename := filepath.Base(path)

     

       256
        
       			isInvalidTest := strings.Contains(filename, "-invalid-")

     

       257
       +
       

     

       258
        
       			if verbose {

     

       259
        
       				if isInvalidTest {

     

       260
        
       					fmt.Printf("\n  Testing (expect failure): %s\n", filename)

     
···

       269
        
       				validationErrors = append(validationErrors, fmt.Sprintf("Failed to open %s: %v", path, err))

     

       270
        
       				return nil

     

       271
        
       			}

     

       272
       +
       			defer func() {

     

       273
       +
       				if closeErr := file.Close(); closeErr != nil {

     

       274
       +
       					validationErrors = append(validationErrors, fmt.Sprintf("Failed to close %s: %v", path, closeErr))

     

       275
       +
       				}

     

       276
       +
       			}()

     

       277
        
       

     

       278
       +
       			data, readErr := io.ReadAll(file)

     

       279
       +
       			if readErr != nil {

     

       280
       +
       				validationErrors = append(validationErrors, fmt.Sprintf("Failed to read %s: %v", path, readErr))

     

       281
        
       				return nil

     

       282
        
       			}

     

       283
        
       

     
···

       285
        
       			var recordData map[string]interface{}

     

       286
        
       			decoder := json.NewDecoder(bytes.NewReader(data))

     

       287
        
       			decoder.UseNumber() // This preserves numbers as json.Number instead of float64

     

       288
       +
       			if decodeErr := decoder.Decode(&recordData); decodeErr != nil {

     

       289
       +
       				validationErrors = append(validationErrors, fmt.Sprintf("Failed to parse JSON in %s: %v", path, decodeErr))

     

       290
        
       				return nil

     

       291
        
       			}

     

       292
       +
       

     

       293
        
       			// Convert json.Number values to appropriate types

     

       294
        
       			recordData = convertNumbers(recordData).(map[string]interface{})

     

       295
        
       

     
···

       309
        
       			}

     

       310
        
       

     

       311
        
       			// Validate the record

     

       312
       +
       			validateErr := lexicon.ValidateRecord(catalog, recordData, recordType, flags)

     

       313
       +
       

     

       314
        
       			if isInvalidTest {

     

       315
        
       				// This file should fail validation

     

       316
        
       				invalidFiles++

     

       317
       +
       				if validateErr != nil {

     

       318
        
       					invalidFailCount++

     

       319
        
       					if verbose {

     

       320
       +
       						fmt.Printf("    ✅ Correctly rejected invalid %s record: %v\n", recordType, validateErr)

     

       321
        
       					}

     

       322
        
       				} else {

     

       323
        
       					validationErrors = append(validationErrors, fmt.Sprintf("Invalid test file %s passed validation when it should have failed", path))

     
···

       328
        
       			} else {

     

       329
        
       				// This file should pass validation

     

       330
        
       				validFiles++

     

       331
       +
       				if validateErr != nil {

     

       332
       +
       					validationErrors = append(validationErrors, fmt.Sprintf("Validation failed for %s (type: %s): %v", path, recordType, validateErr))

     

       333
        
       					if verbose {

     

       334
       +
       						fmt.Printf("    ❌ Failed: %v\n", validateErr)

     

       335
        
       					}

     

       336
        
       				} else {

     

       337
        
       					validSuccessCount++

     
···

       344
        
       		}

     

       345
        
       		return nil

     

       346
        
       	})

     

       0
        
       
     

       347
        
       	if err != nil {

     

       348
        
       		return fmt.Errorf("error walking test data directory: %w", err)

     

       349
        
       	}

     
···

       364
        
       		fmt.Printf("\n📋 Validation Summary:\n")

     

       365
        
       		fmt.Printf("  Valid test files:   %d/%d passed\n", validSuccessCount, validFiles)

     

       366
        
       		fmt.Printf("  Invalid test files: %d/%d correctly rejected\n", invalidFailCount, invalidFiles)

     

       367
       +
       

     

       368
        
       		if validSuccessCount == validFiles && invalidFailCount == invalidFiles {

     

       369
        
       			fmt.Printf("\n  ✅ All test files behaved as expected!\n")

     

       370
        
       		}

     

       371
       +
       

     

       372
        
       		// Show test coverage summary (only for valid files)

     

       373
        
       		fmt.Printf("\n📊 Test Data Coverage Summary:\n")

     

       374
        
       		fmt.Printf("  - Records with test data: %d types\n", len(testedTypes))

     

       375
        
       		fmt.Printf("  - Valid test files: %d\n", validFiles)

     

       376
        
       		fmt.Printf("  - Invalid test files: %d (for error validation)\n", invalidFiles)

     

       377
       +
       

     

       378
        
       		fmt.Printf("\n  Tested record types:\n")

     

       379
        
       		for recordType := range testedTypes {

     

       380
        
       			fmt.Printf("    ✓ %s\n", recordType)

     

       381
        
       		}

     

       382
       +
       

     

       383
        
       		// Show untested schemas

     

       384
        
       		untestedCount := 0

     

       385
        
       		fmt.Printf("\n  ⚠️  Record types without test data:\n")

     
···

       389
        
       				untestedCount++

     

       390
        
       			}

     

       391
        
       		}

     

       392
       +
       

     

       393
        
       		if untestedCount == 0 {

     

       394
        
       			fmt.Println("    (None - full test coverage!)")

     

       395
        
       		} else {

     

       396
       +
       			fmt.Printf("\n  Coverage: %d/%d record types have test data (%.1f%%)\n",

     

       397
       +
       				len(testedTypes), len(allSchemas),

     

       398
        
       				float64(len(testedTypes))/float64(len(allSchemas))*100)

     

       399
        
       		}

     

       400
        
       	}

     
···

       413
        
       		"social.coves.richtext.facet#italic",

     

       414
        
       		"social.coves.richtext.facet#strikethrough",

     

       415
        
       		"social.coves.richtext.facet#spoiler",

     

       416
       +
       

     

       417
        
       		// Post types and views

     

       418
        
       		"social.coves.post.get#postView",

     

       419
        
       		"social.coves.post.get#authorView",

     
···

       423
        
       		"social.coves.post.get#externalView",

     

       424
        
       		"social.coves.post.get#postStats",

     

       425
        
       		"social.coves.post.get#viewerState",

     

       426
       +
       

     

       427
        
       		// Post record types

     

       428
        
       		"social.coves.post.record#originalAuthor",

     

       429
       +
       

     

       430
        
       		// Actor definitions

     

       431
        
       		"social.coves.actor.profile#geoLocation",

     

       432
       +
       

     

       433
        
       		// Community definitions

     

       434
        
       		"social.coves.community.rules#rule",

     

       435
        
       	}

+6 -3

internal/api/handlers/community/create.go

···

       1
        
       package community

     

       2
        
       

     

       3
        
       import (

     

       0
        
       
     

       4
        
       	"encoding/json"

     

       5
        
       	"net/http"

     

       6
       -
       

     

       7
       -
       	"Coves/internal/core/communities"

     

       8
        
       )

     

       9
        
       

     

       10
        
       // CreateHandler handles community creation

     
···

       68
        
       

     

       69
        
       	w.Header().Set("Content-Type", "application/json")

     

       70
        
       	w.WriteHeader(http.StatusOK)

     

       71
       -
       	json.NewEncoder(w).Encode(response)

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       72
        
       }

···

       1
        
       package community

     

       2
        
       

     

       3
        
       import (

     

       4
       +
       	"Coves/internal/core/communities"

     

       5
        
       	"encoding/json"

     

       6
        
       	"net/http"

     

       0
        
       
     

       0
        
       
     

       7
        
       )

     

       8
        
       

     

       9
        
       // CreateHandler handles community creation

     
···

       67
        
       

     

       68
        
       	w.Header().Set("Content-Type", "application/json")

     

       69
        
       	w.WriteHeader(http.StatusOK)

     

       70
       +
       	if err := json.NewEncoder(w).Encode(response); err != nil {

     

       71
       +
       		// Log encoding errors but don't return error response (headers already sent)

     

       72
       +
       		// This follows Go's standard practice for HTTP handlers

     

       73
       +
       		_ = err

     

       74
       +
       	}

     

       75
        
       }

+5 -4

internal/api/handlers/community/errors.go

···

       1
        
       package community

     

       2
        
       

     

       3
        
       import (

     

       0
        
       
     

       4
        
       	"encoding/json"

     

       5
        
       	"log"

     

       6
        
       	"net/http"

     

       7
       -
       

     

       8
       -
       	"Coves/internal/core/communities"

     

       9
        
       )

     

       10
        
       

     

       11
        
       // XRPCError represents an XRPC error response

     
···

       18
        
       func writeError(w http.ResponseWriter, status int, error, message string) {

     

       19
        
       	w.Header().Set("Content-Type", "application/json")

     

       20
        
       	w.WriteHeader(status)

     

       21
       -
       	json.NewEncoder(w).Encode(XRPCError{

     

       22
        
       		Error:   error,

     

       23
        
       		Message: message,

     

       24
       -
       	})

     

       0
        
       
     

       0
        
       
     

       25
        
       }

     

       26
        
       

     

       27
        
       // handleServiceError converts service errors to appropriate HTTP responses

···

       1
        
       package community

     

       2
        
       

     

       3
        
       import (

     

       4
       +
       	"Coves/internal/core/communities"

     

       5
        
       	"encoding/json"

     

       6
        
       	"log"

     

       7
        
       	"net/http"

     

       0
        
       
     

       0
        
       
     

       8
        
       )

     

       9
        
       

     

       10
        
       // XRPCError represents an XRPC error response

     
···

       17
        
       func writeError(w http.ResponseWriter, status int, error, message string) {

     

       18
        
       	w.Header().Set("Content-Type", "application/json")

     

       19
        
       	w.WriteHeader(status)

     

       20
       +
       	if err := json.NewEncoder(w).Encode(XRPCError{

     

       21
        
       		Error:   error,

     

       22
        
       		Message: message,

     

       23
       +
       	}); err != nil {

     

       24
       +
       		log.Printf("Failed to encode error response: %v", err)

     

       25
       +
       	}

     

       26
        
       }

     

       27
        
       

     

       28
        
       // handleServiceError converts service errors to appropriate HTTP responses

+6 -3

internal/api/handlers/community/get.go

···

       1
        
       package community

     

       2
        
       

     

       3
        
       import (

     

       0
        
       
     

       4
        
       	"encoding/json"

     

       5
        
       	"net/http"

     

       6
       -
       

     

       7
       -
       	"Coves/internal/core/communities"

     

       8
        
       )

     

       9
        
       

     

       10
        
       // GetHandler handles community retrieval

     
···

       44
        
       	// Return community data

     

       45
        
       	w.Header().Set("Content-Type", "application/json")

     

       46
        
       	w.WriteHeader(http.StatusOK)

     

       47
       -
       	json.NewEncoder(w).Encode(community)

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       48
        
       }

···

       1
        
       package community

     

       2
        
       

     

       3
        
       import (

     

       4
       +
       	"Coves/internal/core/communities"

     

       5
        
       	"encoding/json"

     

       6
        
       	"net/http"

     

       0
        
       
     

       0
        
       
     

       7
        
       )

     

       8
        
       

     

       9
        
       // GetHandler handles community retrieval

     
···

       43
        
       	// Return community data

     

       44
        
       	w.Header().Set("Content-Type", "application/json")

     

       45
        
       	w.WriteHeader(http.StatusOK)

     

       46
       +
       	if err := json.NewEncoder(w).Encode(community); err != nil {

     

       47
       +
       		// Log encoding errors but don't return error response (headers already sent)

     

       48
       +
       		// This follows Go's standard practice for HTTP handlers

     

       49
       +
       		_ = err

     

       50
       +
       	}

     

       51
        
       }

+6 -3

internal/api/handlers/community/list.go

···

       1
        
       package community

     

       2
        
       

     

       3
        
       import (

     

       0
        
       
     

       4
        
       	"encoding/json"

     

       5
        
       	"net/http"

     

       6
        
       	"strconv"

     

       7
       -
       

     

       8
       -
       	"Coves/internal/core/communities"

     

       9
        
       )

     

       10
        
       

     

       11
        
       // ListHandler handles listing communities

     
···

       70
        
       

     

       71
        
       	w.Header().Set("Content-Type", "application/json")

     

       72
        
       	w.WriteHeader(http.StatusOK)

     

       73
       -
       	json.NewEncoder(w).Encode(response)

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       74
        
       }

···

       1
        
       package community

     

       2
        
       

     

       3
        
       import (

     

       4
       +
       	"Coves/internal/core/communities"

     

       5
        
       	"encoding/json"

     

       6
        
       	"net/http"

     

       7
        
       	"strconv"

     

       0
        
       
     

       0
        
       
     

       8
        
       )

     

       9
        
       

     

       10
        
       // ListHandler handles listing communities

     
···

       69
        
       

     

       70
        
       	w.Header().Set("Content-Type", "application/json")

     

       71
        
       	w.WriteHeader(http.StatusOK)

     

       72
       +
       	if err := json.NewEncoder(w).Encode(response); err != nil {

     

       73
       +
       		// Log encoding errors but don't return error response (headers already sent)

     

       74
       +
       		// This follows Go's standard practice for HTTP handlers

     

       75
       +
       		_ = err

     

       76
       +
       	}

     

       77
        
       }

+6 -3

internal/api/handlers/community/search.go

···

       1
        
       package community

     

       2
        
       

     

       3
        
       import (

     

       0
        
       
     

       4
        
       	"encoding/json"

     

       5
        
       	"net/http"

     

       6
        
       	"strconv"

     

       7
       -
       

     

       8
       -
       	"Coves/internal/core/communities"

     

       9
        
       )

     

       10
        
       

     

       11
        
       // SearchHandler handles community search

     
···

       74
        
       

     

       75
        
       	w.Header().Set("Content-Type", "application/json")

     

       76
        
       	w.WriteHeader(http.StatusOK)

     

       77
       -
       	json.NewEncoder(w).Encode(response)

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       78
        
       }

···

       1
        
       package community

     

       2
        
       

     

       3
        
       import (

     

       4
       +
       	"Coves/internal/core/communities"

     

       5
        
       	"encoding/json"

     

       6
        
       	"net/http"

     

       7
        
       	"strconv"

     

       0
        
       
     

       0
        
       
     

       8
        
       )

     

       9
        
       

     

       10
        
       // SearchHandler handles community search

     
···

       73
        
       

     

       74
        
       	w.Header().Set("Content-Type", "application/json")

     

       75
        
       	w.WriteHeader(http.StatusOK)

     

       76
       +
       	if err := json.NewEncoder(w).Encode(response); err != nil {

     

       77
       +
       		// Log encoding errors but don't return error response (headers already sent)

     

       78
       +
       		// This follows Go's standard practice for HTTP handlers

     

       79
       +
       		_ = err

     

       80
       +
       	}

     

       81
        
       }

+9 -4

internal/api/handlers/oauth/callback.go

···

       1
        
       package oauth

     

       2
        
       

     

       3
        
       import (

     

       0
        
       
     

       4
        
       	"log"

     

       5
        
       	"net/http"

     

       6
        
       	"os"

     

       7
        
       	"strings"

     

       8
        
       	"time"

     

       9
        
       

     

       10
       -
       	"Coves/internal/atproto/oauth"

     

       11
        
       	oauthCore "Coves/internal/core/oauth"

     

       12
        
       )

     

       13
        
       

     
···

       154
        
       		ExpiresAt:           expiresAt,

     

       155
        
       	}

     

       156
        
       

     

       157
       -
       	if err := h.sessionStore.SaveSession(session); err != nil {

     

       158
       -
       		log.Printf("Failed to save OAuth session: %v", err)

     

       159
        
       		http.Error(w, "Failed to save session", http.StatusInternalServerError)

     

       160
        
       		return

     

       161
        
       	}

     
···

       168
        
       	if err != nil {

     

       169
        
       		log.Printf("Failed to get cookie session: %v", err)

     

       170
        
       		// Try to create a new session anyway

     

       171
       -
       		httpSession, _ = cookieStore.New(r, sessionName)

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       172
        
       	}

     

       173
        
       

     

       174
        
       	httpSession.Values[sessionDID] = oauthReq.DID

···

       1
        
       package oauth

     

       2
        
       

     

       3
        
       import (

     

       4
       +
       	"Coves/internal/atproto/oauth"

     

       5
        
       	"log"

     

       6
        
       	"net/http"

     

       7
        
       	"os"

     

       8
        
       	"strings"

     

       9
        
       	"time"

     

       10
        
       

     

       0
        
       
     

       11
        
       	oauthCore "Coves/internal/core/oauth"

     

       12
        
       )

     

       13
        
       

     
···

       154
        
       		ExpiresAt:           expiresAt,

     

       155
        
       	}

     

       156
        
       

     

       157
       +
       	if saveErr := h.sessionStore.SaveSession(session); saveErr != nil {

     

       158
       +
       		log.Printf("Failed to save OAuth session: %v", saveErr)

     

       159
        
       		http.Error(w, "Failed to save session", http.StatusInternalServerError)

     

       160
        
       		return

     

       161
        
       	}

     
···

       168
        
       	if err != nil {

     

       169
        
       		log.Printf("Failed to get cookie session: %v", err)

     

       170
        
       		// Try to create a new session anyway

     

       171
       +
       		httpSession, err = cookieStore.New(r, sessionName)

     

       172
       +
       		if err != nil {

     

       173
       +
       			log.Printf("Failed to create new session: %v", err)

     

       174
       +
       			http.Error(w, "Failed to create session", http.StatusInternalServerError)

     

       175
       +
       			return

     

       176
       +
       		}

     

       177
        
       	}

     

       178
        
       

     

       179
        
       	httpSession.Values[sessionDID] = oauthReq.DID

+16 -4

internal/api/handlers/oauth/env_test.go

···

       62
        
       		t.Run(tt.name, func(t *testing.T) {

     

       63
        
       			// Set environment variable

     

       64
        
       			if tt.envValue != "" {

     

       65
       -
       				os.Setenv(tt.envKey, tt.envValue)

     

       66
       -
       				defer os.Unsetenv(tt.envKey)

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       67
        
       			}

     

       68
        
       

     

       69
        
       			got, err := GetEnvBase64OrPlain(tt.envKey)

     
···

       103
        
       

     

       104
        
       	for _, tt := range tests {

     

       105
        
       		t.Run(tt.name, func(t *testing.T) {

     

       106
       -
       			os.Setenv("TEST_REAL_JWK", tt.envValue)

     

       107
       -
       			defer os.Unsetenv("TEST_REAL_JWK")

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       108
        
       

     

       109
        
       			got, err := GetEnvBase64OrPlain("TEST_REAL_JWK")

     

       110
        
       			if err != nil {

···

       62
        
       		t.Run(tt.name, func(t *testing.T) {

     

       63
        
       			// Set environment variable

     

       64
        
       			if tt.envValue != "" {

     

       65
       +
       				if err := os.Setenv(tt.envKey, tt.envValue); err != nil {

     

       66
       +
       					t.Fatalf("Failed to set env var: %v", err)

     

       67
       +
       				}

     

       68
       +
       				defer func() {

     

       69
       +
       					if err := os.Unsetenv(tt.envKey); err != nil {

     

       70
       +
       						t.Errorf("Failed to unset env var: %v", err)

     

       71
       +
       					}

     

       72
       +
       				}()

     

       73
        
       			}

     

       74
        
       

     

       75
        
       			got, err := GetEnvBase64OrPlain(tt.envKey)

     
···

       109
        
       

     

       110
        
       	for _, tt := range tests {

     

       111
        
       		t.Run(tt.name, func(t *testing.T) {

     

       112
       +
       			if err := os.Setenv("TEST_REAL_JWK", tt.envValue); err != nil {

     

       113
       +
       				t.Fatalf("Failed to set env var: %v", err)

     

       114
       +
       			}

     

       115
       +
       			defer func() {

     

       116
       +
       				if err := os.Unsetenv("TEST_REAL_JWK"); err != nil {

     

       117
       +
       					t.Errorf("Failed to unset env var: %v", err)

     

       118
       +
       				}

     

       119
       +
       			}()

     

       120
        
       

     

       121
        
       			got, err := GetEnvBase64OrPlain("TEST_REAL_JWK")

     

       122
        
       			if err != nil {

+5 -3

internal/api/handlers/oauth/jwks.go

···

       1
        
       package oauth

     

       2
        
       

     

       3
        
       import (

     

       0
        
       
     

       4
        
       	"encoding/json"

     

       0
        
       
     

       5
        
       	"net/http"

     

       6
       -
       

     

       7
       -
       	"Coves/internal/atproto/oauth"

     

       8
        
       

     

       9
        
       	"github.com/lestrrat-go/jwx/v2/jwk"

     

       10
        
       )

     
···

       47
        
       	// Serve JWKS

     

       48
        
       	w.Header().Set("Content-Type", "application/json")

     

       49
        
       	w.WriteHeader(http.StatusOK)

     

       50
       -
       	json.NewEncoder(w).Encode(jwks)

     

       0
        
       
     

       0
        
       
     

       51
        
       }

···

       1
        
       package oauth

     

       2
        
       

     

       3
        
       import (

     

       4
       +
       	"Coves/internal/atproto/oauth"

     

       5
        
       	"encoding/json"

     

       6
       +
       	"log"

     

       7
        
       	"net/http"

     

       0
        
       
     

       0
        
       
     

       8
        
       

     

       9
        
       	"github.com/lestrrat-go/jwx/v2/jwk"

     

       10
        
       )

     
···

       47
        
       	// Serve JWKS

     

       48
        
       	w.Header().Set("Content-Type", "application/json")

     

       49
        
       	w.WriteHeader(http.StatusOK)

     

       50
       +
       	if err := json.NewEncoder(w).Encode(jwks); err != nil {

     

       51
       +
       		log.Printf("Failed to encode JWKS response: %v", err)

     

       52
       +
       	}

     

       53
        
       }

+7 -5

internal/api/handlers/oauth/login.go

···

       1
        
       package oauth

     

       2
        
       

     

       3
        
       import (

     

       0
        
       
     

       0
        
       
     

       4
        
       	"encoding/json"

     

       5
        
       	"log"

     

       6
        
       	"net/http"

     

       7
        
       	"net/url"

     

       8
        
       	"strings"

     

       9
        
       

     

       10
       -
       	"Coves/internal/atproto/identity"

     

       11
       -
       	"Coves/internal/atproto/oauth"

     

       12
        
       	oauthCore "Coves/internal/core/oauth"

     

       13
        
       )

     

       14
        
       

     
···

       144
        
       		ReturnURL:           req.ReturnURL,

     

       145
        
       	}

     

       146
        
       

     

       147
       -
       	if err := h.sessionStore.SaveRequest(oauthReq); err != nil {

     

       148
       -
       		log.Printf("Failed to save OAuth request: %v", err)

     

       149
        
       		http.Error(w, "Failed to save authorization state", http.StatusInternalServerError)

     

       150
        
       		return

     

       151
        
       	}

     
···

       171
        
       

     

       172
        
       	w.Header().Set("Content-Type", "application/json")

     

       173
        
       	w.WriteHeader(http.StatusOK)

     

       174
       -
       	json.NewEncoder(w).Encode(resp)

     

       0
        
       
     

       0
        
       
     

       175
        
       }

···

       1
        
       package oauth

     

       2
        
       

     

       3
        
       import (

     

       4
       +
       	"Coves/internal/atproto/identity"

     

       5
       +
       	"Coves/internal/atproto/oauth"

     

       6
        
       	"encoding/json"

     

       7
        
       	"log"

     

       8
        
       	"net/http"

     

       9
        
       	"net/url"

     

       10
        
       	"strings"

     

       11
        
       

     

       0
        
       
     

       0
        
       
     

       12
        
       	oauthCore "Coves/internal/core/oauth"

     

       13
        
       )

     

       14
        
       

     
···

       144
        
       		ReturnURL:           req.ReturnURL,

     

       145
        
       	}

     

       146
        
       

     

       147
       +
       	if saveErr := h.sessionStore.SaveRequest(oauthReq); saveErr != nil {

     

       148
       +
       		log.Printf("Failed to save OAuth request: %v", saveErr)

     

       149
        
       		http.Error(w, "Failed to save authorization state", http.StatusInternalServerError)

     

       150
        
       		return

     

       151
        
       	}

     
···

       171
        
       

     

       172
        
       	w.Header().Set("Content-Type", "application/json")

     

       173
        
       	w.WriteHeader(http.StatusOK)

     

       174
       +
       	if err := json.NewEncoder(w).Encode(resp); err != nil {

     

       175
       +
       		log.Printf("Failed to encode response: %v", err)

     

       176
       +
       	}

     

       177
        
       }

+10 -6

internal/api/handlers/oauth/metadata.go

···

       13
        
       	ClientID                    string   `json:"client_id"`

     

       14
        
       	ClientName                  string   `json:"client_name"`

     

       15
        
       	ClientURI                   string   `json:"client_uri"`

     

       16
       -
       	RedirectURIs                []string `json:"redirect_uris"`

     

       17
       -
       	GrantTypes                  []string `json:"grant_types"`

     

       18
       -
       	ResponseTypes               []string `json:"response_types"`

     

       19
        
       	Scope                       string   `json:"scope"`

     

       20
        
       	TokenEndpointAuthMethod     string   `json:"token_endpoint_auth_method"`

     

       21
        
       	TokenEndpointAuthSigningAlg string   `json:"token_endpoint_auth_signing_alg"`

     

       22
       -
       	DpopBoundAccessTokens       bool     `json:"dpop_bound_access_tokens"`

     

       23
        
       	ApplicationType             string   `json:"application_type"`

     

       24
       -
       	JwksURI                     string   `json:"jwks_uri,omitempty"` // Only in production

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       25
        
       }

     

       26
        
       

     

       27
        
       // HandleClientMetadata serves the OAuth client metadata

     
···

       55
        
       

     

       56
        
       	w.Header().Set("Content-Type", "application/json")

     

       57
        
       	w.WriteHeader(http.StatusOK)

     

       58
       -
       	json.NewEncoder(w).Encode(metadata)

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       59
        
       }

     

       60
        
       

     

       61
        
       // getAppViewURL returns the public URL of the AppView

···

       13
        
       	ClientID                    string   `json:"client_id"`

     

       14
        
       	ClientName                  string   `json:"client_name"`

     

       15
        
       	ClientURI                   string   `json:"client_uri"`

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       16
        
       	Scope                       string   `json:"scope"`

     

       17
        
       	TokenEndpointAuthMethod     string   `json:"token_endpoint_auth_method"`

     

       18
        
       	TokenEndpointAuthSigningAlg string   `json:"token_endpoint_auth_signing_alg"`

     

       0
        
       
     

       19
        
       	ApplicationType             string   `json:"application_type"`

     

       20
       +
       	JwksURI                     string   `json:"jwks_uri,omitempty"`

     

       21
       +
       	RedirectURIs                []string `json:"redirect_uris"`

     

       22
       +
       	GrantTypes                  []string `json:"grant_types"`

     

       23
       +
       	ResponseTypes               []string `json:"response_types"`

     

       24
       +
       	DpopBoundAccessTokens       bool     `json:"dpop_bound_access_tokens"`

     

       25
        
       }

     

       26
        
       

     

       27
        
       // HandleClientMetadata serves the OAuth client metadata

     
···

       55
        
       

     

       56
        
       	w.Header().Set("Content-Type", "application/json")

     

       57
        
       	w.WriteHeader(http.StatusOK)

     

       58
       +
       	if err := json.NewEncoder(w).Encode(metadata); err != nil {

     

       59
       +
       		// Log encoding errors but don't return error response (headers already sent)

     

       60
       +
       		// This follows Go's standard practice for HTTP handlers

     

       61
       +
       		_ = err

     

       62
       +
       	}

     

       63
        
       }

     

       64
        
       

     

       65
        
       // getAppViewURL returns the public URL of the AppView

+13 -6

internal/api/routes/user.go

···

       1
        
       package routes

     

       2
        
       

     

       3
        
       import (

     

       0
        
       
     

       4
        
       	"encoding/json"

     

       5
        
       	"errors"

     

       0
        
       
     

       6
        
       	"net/http"

     

       7
        
       	"time"

     

       8
        
       

     

       9
       -
       	"Coves/internal/core/users"

     

       10
        
       	"github.com/go-chi/chi/v5"

     

       11
        
       )

     

       12
        
       

     
···

       73
        
       

     

       74
        
       	w.Header().Set("Content-Type", "application/json")

     

       75
        
       	w.WriteHeader(http.StatusOK)

     

       76
       -
       	json.NewEncoder(w).Encode(response)

     

       0
        
       
     

       0
        
       
     

       77
        
       }

     

       78
        
       

     

       79
        
       // Signup handles social.coves.actor.signup

     
···

       106
        
       

     

       107
        
       	w.Header().Set("Content-Type", "application/json")

     

       108
        
       	w.WriteHeader(http.StatusOK)

     

       109
       -
       	json.NewEncoder(w).Encode(response)

     

       0
        
       
     

       0
        
       
     

       110
        
       }

     

       111
        
       

     

       112
        
       // respondWithLexiconError maps domain errors to lexicon error types and HTTP status codes

     
···

       168
        
       	// XRPC error response format

     

       169
        
       	w.Header().Set("Content-Type", "application/json")

     

       170
        
       	w.WriteHeader(statusCode)

     

       171
       -
       	json.NewEncoder(w).Encode(map[string]interface{}{

     

       172
        
       		"error":   errorName,

     

       173
        
       		"message": message,

     

       174
       -
       	})

     

       175
       -
       }
     

       0
        
       
     

       0

···

       1
        
       package routes

     

       2
        
       

     

       3
        
       import (

     

       4
       +
       	"Coves/internal/core/users"

     

       5
        
       	"encoding/json"

     

       6
        
       	"errors"

     

       7
       +
       	"log"

     

       8
        
       	"net/http"

     

       9
        
       	"time"

     

       10
        
       

     

       0
        
       
     

       11
        
       	"github.com/go-chi/chi/v5"

     

       12
        
       )

     

       13
        
       

     
···

       74
        
       

     

       75
        
       	w.Header().Set("Content-Type", "application/json")

     

       76
        
       	w.WriteHeader(http.StatusOK)

     

       77
       +
       	if err := json.NewEncoder(w).Encode(response); err != nil {

     

       78
       +
       		log.Printf("Failed to encode response: %v", err)

     

       79
       +
       	}

     

       80
        
       }

     

       81
        
       

     

       82
        
       // Signup handles social.coves.actor.signup

     
···

       109
        
       

     

       110
        
       	w.Header().Set("Content-Type", "application/json")

     

       111
        
       	w.WriteHeader(http.StatusOK)

     

       112
       +
       	if err := json.NewEncoder(w).Encode(response); err != nil {

     

       113
       +
       		log.Printf("Failed to encode response: %v", err)

     

       114
       +
       	}

     

       115
        
       }

     

       116
        
       

     

       117
        
       // respondWithLexiconError maps domain errors to lexicon error types and HTTP status codes

     
···

       173
        
       	// XRPC error response format

     

       174
        
       	w.Header().Set("Content-Type", "application/json")

     

       175
        
       	w.WriteHeader(statusCode)

     

       176
       +
       	if err := json.NewEncoder(w).Encode(map[string]interface{}{

     

       177
        
       		"error":   errorName,

     

       178
        
       		"message": message,

     

       179
       +
       	}); err != nil {

     

       180
       +
       		log.Printf("Failed to encode error response: %v", err)

     

       181
       +
       	}

     

       182
       +
       }

+7 -6

internal/atproto/did/generator.go

···

       9
        
       

     

       10
        
       // Generator creates DIDs for Coves entities

     

       11
        
       type Generator struct {

     

       12
       -
       	isDevEnv        bool   // true = generate without registering, false = register with PLC

     

       13
       -
       	plcDirectoryURL string // PLC directory URL (only used when isDevEnv=false)

     

       14
        
       }

     

       15
        
       

     

       16
        
       // NewGenerator creates a new DID generator

     
···

       51
        
       	// 4. Store keypair securely for future DID updates

     

       52
        
       	//

     

       53
        
       	// For now, we just generate the identifier (works fine for local dev)

     

       54
       -
       	if !g.isDevEnv {

     

       55
       -
       		// Future: implement PLC registration here

     

       56
       -
       		// return "", fmt.Errorf("PLC registration not yet implemented")

     

       57
       -
       	}

     

       0
        
       
     

       58
        
       

     

       59
        
       	return did, nil

     

       60
        
       }

···

       9
        
       

     

       10
        
       // Generator creates DIDs for Coves entities

     

       11
        
       type Generator struct {

     

       12
       +
       	plcDirectoryURL string

     

       13
       +
       	isDevEnv        bool

     

       14
        
       }

     

       15
        
       

     

       16
        
       // NewGenerator creates a new DID generator

     
···

       51
        
       	// 4. Store keypair securely for future DID updates

     

       52
        
       	//

     

       53
        
       	// For now, we just generate the identifier (works fine for local dev)

     

       54
       +
       	// Production PLC registration is not yet implemented - DIDs are generated

     

       55
       +
       	// locally but not registered with the PLC directory. This is acceptable

     

       56
       +
       	// for development and private instances, but production deployments should

     

       57
       +
       	// implement full PLC registration to ensure DIDs are globally resolvable.

     

       58
       +
       	_ = g.isDevEnv // Acknowledge that isDevEnv will be used when PLC registration is implemented

     

       59
        
       

     

       60
        
       	return did, nil

     

       61
        
       }

+2 -2

internal/atproto/identity/postgres_cache.go

···

       143
        
       		return fmt.Errorf("failed to purge identity cache: %w", err)

     

       144
        
       	}

     

       145
        
       

     

       146
       -
       	rowsAffected, _ := result.RowsAffected()

     

       147
       -
       	if rowsAffected > 0 {

     

       148
        
       		log.Printf("[identity-cache] Purged %d entries for: %s", rowsAffected, identifier)

     

       149
        
       	}

     

       150

···

       143
        
       		return fmt.Errorf("failed to purge identity cache: %w", err)

     

       144
        
       	}

     

       145
        
       

     

       146
       +
       	rowsAffected, err := result.RowsAffected()

     

       147
       +
       	if err == nil && rowsAffected > 0 {

     

       148
        
       		log.Printf("[identity-cache] Purged %d entries for: %s", rowsAffected, identifier)

     

       149
        
       	}

     

       150

+21 -39

internal/atproto/jetstream/community_consumer.go

···

       1
        
       package jetstream

     

       2
        
       

     

       3
        
       import (

     

       0
        
       
     

       4
        
       	"context"

     

       5
        
       	"encoding/json"

     

       6
        
       	"fmt"

     

       7
        
       	"log"

     

       8
       -
       	"strings"

     

       9
        
       	"time"

     

       10
       -
       

     

       11
       -
       	"Coves/internal/core/communities"

     

       12
        
       )

     

       13
        
       

     

       14
        
       // CommunityEventConsumer consumes community-related events from Jetstream

     
···

       124
        
       

     

       125
        
       	// Handle description facets (rich text)

     

       126
        
       	if profile.DescriptionFacets != nil {

     

       127
       -
       		facetsJSON, err := json.Marshal(profile.DescriptionFacets)

     

       128
       -
       		if err == nil {

     

       129
        
       			community.DescriptionFacets = facetsJSON

     

       130
        
       		}

     

       131
        
       	}

     
···

       195
        
       

     

       196
        
       	// Update description facets

     

       197
        
       	if profile.DescriptionFacets != nil {

     

       198
       -
       		facetsJSON, err := json.Marshal(profile.DescriptionFacets)

     

       199
       -
       		if err == nil {

     

       200
        
       			existing.DescriptionFacets = facetsJSON

     

       201
        
       		}

     

       202
        
       	}

     
···

       296
        
       // Helper types and functions

     

       297
        
       

     

       298
        
       type CommunityProfile struct {

     

       299
       -
       	// V2 ONLY: No DID field (repo DID is authoritative)

     

       300
       -
       	Handle            string                 `json:"handle"`        // Scoped handle (!gaming@coves.social)

     

       301
       -
       	AtprotoHandle     string                 `json:"atprotoHandle"` // Real atProto handle (gaming.communities.coves.social)

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       302
        
       	Name              string                 `json:"name"`

     

       303
       -
       	DisplayName       string                 `json:"displayName"`

     

       0
        
       
     

       304
        
       	Description       string                 `json:"description"`

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       305
        
       	DescriptionFacets []interface{}          `json:"descriptionFacets"`

     

       306
       -
       	Avatar            map[string]interface{} `json:"avatar"`

     

       307
       -
       	Banner            map[string]interface{} `json:"banner"`

     

       308
       -
       	// Owner field removed - V2 communities ALWAYS self-own (owner == repo DID)

     

       309
       -
       	CreatedBy       string           `json:"createdBy"`

     

       310
       -
       	HostedBy        string           `json:"hostedBy"`

     

       311
       -
       	Visibility      string           `json:"visibility"`

     

       312
       -
       	Federation      FederationConfig `json:"federation"`

     

       313
       -
       	ModerationType  string           `json:"moderationType"`

     

       314
       -
       	ContentWarnings []string         `json:"contentWarnings"`

     

       315
       -
       	MemberCount     int              `json:"memberCount"`

     

       316
       -
       	SubscriberCount int              `json:"subscriberCount"`

     

       317
       -
       	FederatedFrom   string           `json:"federatedFrom"`

     

       318
       -
       	FederatedID     string           `json:"federatedId"`

     

       319
       -
       	CreatedAt       time.Time        `json:"createdAt"`

     

       320
        
       }

     

       321
        
       

     

       322
        
       type FederationConfig struct {

     
···

       365
        
       

     

       366
        
       	return link, true

     

       367
        
       }

     

       368
       -
       

     

       369
       -
       // validateHandle checks if a handle matches expected format (!name@instance)

     

       370
       -
       func validateHandle(handle string) bool {

     

       371
       -
       	if !strings.HasPrefix(handle, "!") {

     

       372
       -
       		return false

     

       373
       -
       	}

     

       374
       -
       

     

       375
       -
       	parts := strings.Split(handle, "@")

     

       376
       -
       	if len(parts) != 2 {

     

       377
       -
       		return false

     

       378
       -
       	}

     

       379
       -
       

     

       380
       -
       	return true

     

       381
       -
       }

···

       1
        
       package jetstream

     

       2
        
       

     

       3
        
       import (

     

       4
       +
       	"Coves/internal/core/communities"

     

       5
        
       	"context"

     

       6
        
       	"encoding/json"

     

       7
        
       	"fmt"

     

       8
        
       	"log"

     

       0
        
       
     

       9
        
       	"time"

     

       0
        
       
     

       0
        
       
     

       10
        
       )

     

       11
        
       

     

       12
        
       // CommunityEventConsumer consumes community-related events from Jetstream

     
···

       122
        
       

     

       123
        
       	// Handle description facets (rich text)

     

       124
        
       	if profile.DescriptionFacets != nil {

     

       125
       +
       		facetsJSON, marshalErr := json.Marshal(profile.DescriptionFacets)

     

       126
       +
       		if marshalErr == nil {

     

       127
        
       			community.DescriptionFacets = facetsJSON

     

       128
        
       		}

     

       129
        
       	}

     
···

       193
        
       

     

       194
        
       	// Update description facets

     

       195
        
       	if profile.DescriptionFacets != nil {

     

       196
       +
       		facetsJSON, marshalErr := json.Marshal(profile.DescriptionFacets)

     

       197
       +
       		if marshalErr == nil {

     

       198
        
       			existing.DescriptionFacets = facetsJSON

     

       199
        
       		}

     

       200
        
       	}

     
···

       294
        
       // Helper types and functions

     

       295
        
       

     

       296
        
       type CommunityProfile struct {

     

       297
       +
       	CreatedAt         time.Time              `json:"createdAt"`

     

       298
       +
       	Avatar            map[string]interface{} `json:"avatar"`

     

       299
       +
       	Banner            map[string]interface{} `json:"banner"`

     

       300
       +
       	CreatedBy         string                 `json:"createdBy"`

     

       301
       +
       	Visibility        string                 `json:"visibility"`

     

       302
       +
       	AtprotoHandle     string                 `json:"atprotoHandle"`

     

       303
       +
       	DisplayName       string                 `json:"displayName"`

     

       304
        
       	Name              string                 `json:"name"`

     

       305
       +
       	Handle            string                 `json:"handle"`

     

       306
       +
       	HostedBy          string                 `json:"hostedBy"`

     

       307
        
       	Description       string                 `json:"description"`

     

       308
       +
       	FederatedID       string                 `json:"federatedId"`

     

       309
       +
       	ModerationType    string                 `json:"moderationType"`

     

       310
       +
       	FederatedFrom     string                 `json:"federatedFrom"`

     

       311
       +
       	ContentWarnings   []string               `json:"contentWarnings"`

     

       312
        
       	DescriptionFacets []interface{}          `json:"descriptionFacets"`

     

       313
       +
       	MemberCount       int                    `json:"memberCount"`

     

       314
       +
       	SubscriberCount   int                    `json:"subscriberCount"`

     

       315
       +
       	Federation        FederationConfig       `json:"federation"`

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       316
        
       }

     

       317
        
       

     

       318
        
       type FederationConfig struct {

     
···

       361
        
       

     

       362
        
       	return link, true

     

       363
        
       }

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0

+23 -13

internal/atproto/jetstream/user_consumer.go

···

       1
        
       package jetstream

     

       2
        
       

     

       3
        
       import (

     

       0
        
       
     

       0
        
       
     

       4
        
       	"context"

     

       5
        
       	"encoding/json"

     

       6
        
       	"fmt"

     

       7
        
       	"log"

     

       8
        
       	"time"

     

       9
        
       

     

       10
       -
       	"Coves/internal/atproto/identity"

     

       11
       -
       	"Coves/internal/core/users"

     

       12
        
       	"github.com/gorilla/websocket"

     

       13
        
       )

     

       14
        
       

     

       15
        
       // JetstreamEvent represents an event from the Jetstream firehose

     

       16
        
       // Jetstream documentation: https://docs.bsky.app/docs/advanced-guides/jetstream

     

       17
        
       type JetstreamEvent struct {

     

       18
       -
       	Did      string         `json:"did"`

     

       19
       -
       	TimeUS   int64          `json:"time_us"`

     

       20
       -
       	Kind     string         `json:"kind"` // "account", "commit", "identity"

     

       21
        
       	Account  *AccountEvent  `json:"account,omitempty"`

     

       22
        
       	Identity *IdentityEvent `json:"identity,omitempty"`

     

       23
        
       	Commit   *CommitEvent   `json:"commit,omitempty"`

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       24
        
       }

     

       25
        
       

     

       26
        
       type AccountEvent struct {

     

       27
       -
       	Active bool   `json:"active"`

     

       28
        
       	Did    string `json:"did"`

     

       0
        
       
     

       29
        
       	Seq    int64  `json:"seq"`

     

       30
       -
       	Time   string `json:"time"`

     

       31
        
       }

     

       32
        
       

     

       33
        
       type IdentityEvent struct {

     

       34
        
       	Did    string `json:"did"`

     

       35
        
       	Handle string `json:"handle"`

     

       36
       -
       	Seq    int64  `json:"seq"`

     

       37
        
       	Time   string `json:"time"`

     

       0
        
       
     

       38
        
       }

     

       39
        
       

     

       40
        
       // CommitEvent represents a record commit from Jetstream

     
···

       56
        
       }

     

       57
        
       

     

       58
        
       // NewUserEventConsumer creates a new Jetstream consumer for user events

     

       59
       -
       func NewUserEventConsumer(userService users.UserService, identityResolver identity.Resolver, wsURL string, pdsFilter string) *UserEventConsumer {

     

       60
        
       	return &UserEventConsumer{

     

       61
        
       		userService:      userService,

     

       62
        
       		identityResolver: identityResolver,

     
···

       91
        
       	if err != nil {

     

       92
        
       		return fmt.Errorf("failed to connect to Jetstream: %w", err)

     

       93
        
       	}

     

       94
       -
       	defer conn.Close()

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       95
        
       

     

       96
        
       	log.Println("Connected to Jetstream")

     

       97
        
       

     

       98
        
       	// Set read deadline to detect connection issues

     

       99
       -
       	conn.SetReadDeadline(time.Now().Add(60 * time.Second))

     

       0
        
       
     

       0
        
       
     

       100
        
       

     

       101
        
       	// Set pong handler to keep connection alive

     

       102
        
       	conn.SetPongHandler(func(string) error {

     

       103
       -
       		conn.SetReadDeadline(time.Now().Add(60 * time.Second))

     

       0
        
       
     

       0
        
       
     

       104
        
       		return nil

     

       105
        
       	})

     

       106
        
       

     
···

       146
        
       			}

     

       147
        
       

     

       148
        
       			// Reset read deadline on successful read

     

       149
       -
       			conn.SetReadDeadline(time.Now().Add(60 * time.Second))

     

       0
        
       
     

       0
        
       
     

       150
        
       

     

       151
        
       			if err := c.handleEvent(ctx, message); err != nil {

     

       152
        
       				log.Printf("Error handling event: %v", err)

···

       1
        
       package jetstream

     

       2
        
       

     

       3
        
       import (

     

       4
       +
       	"Coves/internal/atproto/identity"

     

       5
       +
       	"Coves/internal/core/users"

     

       6
        
       	"context"

     

       7
        
       	"encoding/json"

     

       8
        
       	"fmt"

     

       9
        
       	"log"

     

       10
        
       	"time"

     

       11
        
       

     

       0
        
       
     

       0
        
       
     

       12
        
       	"github.com/gorilla/websocket"

     

       13
        
       )

     

       14
        
       

     

       15
        
       // JetstreamEvent represents an event from the Jetstream firehose

     

       16
        
       // Jetstream documentation: https://docs.bsky.app/docs/advanced-guides/jetstream

     

       17
        
       type JetstreamEvent struct {

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       18
        
       	Account  *AccountEvent  `json:"account,omitempty"`

     

       19
        
       	Identity *IdentityEvent `json:"identity,omitempty"`

     

       20
        
       	Commit   *CommitEvent   `json:"commit,omitempty"`

     

       21
       +
       	Did      string         `json:"did"`

     

       22
       +
       	Kind     string         `json:"kind"`

     

       23
       +
       	TimeUS   int64          `json:"time_us"`

     

       24
        
       }

     

       25
        
       

     

       26
        
       type AccountEvent struct {

     

       0
        
       
     

       27
        
       	Did    string `json:"did"`

     

       28
       +
       	Time   string `json:"time"`

     

       29
        
       	Seq    int64  `json:"seq"`

     

       30
       +
       	Active bool   `json:"active"`

     

       31
        
       }

     

       32
        
       

     

       33
        
       type IdentityEvent struct {

     

       34
        
       	Did    string `json:"did"`

     

       35
        
       	Handle string `json:"handle"`

     

       0
        
       
     

       36
        
       	Time   string `json:"time"`

     

       37
       +
       	Seq    int64  `json:"seq"`

     

       38
        
       }

     

       39
        
       

     

       40
        
       // CommitEvent represents a record commit from Jetstream

     
···

       56
        
       }

     

       57
        
       

     

       58
        
       // NewUserEventConsumer creates a new Jetstream consumer for user events

     

       59
       +
       func NewUserEventConsumer(userService users.UserService, identityResolver identity.Resolver, wsURL, pdsFilter string) *UserEventConsumer {

     

       60
        
       	return &UserEventConsumer{

     

       61
        
       		userService:      userService,

     

       62
        
       		identityResolver: identityResolver,

     
···

       91
        
       	if err != nil {

     

       92
        
       		return fmt.Errorf("failed to connect to Jetstream: %w", err)

     

       93
        
       	}

     

       94
       +
       	defer func() {

     

       95
       +
       		if err := conn.Close(); err != nil {

     

       96
       +
       			log.Printf("Failed to close WebSocket connection: %v", err)

     

       97
       +
       		}

     

       98
       +
       	}()

     

       99
        
       

     

       100
        
       	log.Println("Connected to Jetstream")

     

       101
        
       

     

       102
        
       	// Set read deadline to detect connection issues

     

       103
       +
       	if err := conn.SetReadDeadline(time.Now().Add(60 * time.Second)); err != nil {

     

       104
       +
       		log.Printf("Failed to set read deadline: %v", err)

     

       105
       +
       	}

     

       106
        
       

     

       107
        
       	// Set pong handler to keep connection alive

     

       108
        
       	conn.SetPongHandler(func(string) error {

     

       109
       +
       		if err := conn.SetReadDeadline(time.Now().Add(60 * time.Second)); err != nil {

     

       110
       +
       			log.Printf("Failed to set read deadline in pong handler: %v", err)

     

       111
       +
       		}

     

       112
        
       		return nil

     

       113
        
       	})

     

       114
        
       

     
···

       154
        
       			}

     

       155
        
       

     

       156
        
       			// Reset read deadline on successful read

     

       157
       +
       			if err := conn.SetReadDeadline(time.Now().Add(60 * time.Second)); err != nil {

     

       158
       +
       				log.Printf("Failed to set read deadline: %v", err)

     

       159
       +
       			}

     

       160
        
       

     

       161
        
       			if err := c.handleEvent(ctx, message); err != nil {

     

       162
        
       				log.Printf("Error handling event: %v", err)

+6 -6

internal/atproto/oauth/dpop.go

···

       92
        
       	// Create headers with DPoP-specific fields

     

       93
        
       	// RFC 9449 requires the "jwk" header to contain the public key as a JSON object

     

       94
        
       	headers := jws.NewHeaders()

     

       95
       -
       	if err := headers.Set(jws.AlgorithmKey, jwa.ES256); err != nil {

     

       96
       -
       		return "", fmt.Errorf("failed to set algorithm: %w", err)

     

       97
        
       	}

     

       98
       -
       	if err := headers.Set(jws.TypeKey, "dpop+jwt"); err != nil {

     

       99
       -
       		return "", fmt.Errorf("failed to set type: %w", err)

     

       100
        
       	}

     

       101
        
       	// Set the public JWK directly - jwx library will handle serialization

     

       102
       -
       	if err := headers.Set(jws.JWKKey, pubKey); err != nil {

     

       103
       -
       		return "", fmt.Errorf("failed to set JWK: %w", err)

     

       104
        
       	}

     

       105
        
       

     

       106
        
       	// Sign using jws.Sign to preserve custom headers

···

       92
        
       	// Create headers with DPoP-specific fields

     

       93
        
       	// RFC 9449 requires the "jwk" header to contain the public key as a JSON object

     

       94
        
       	headers := jws.NewHeaders()

     

       95
       +
       	if setErr := headers.Set(jws.AlgorithmKey, jwa.ES256); setErr != nil {

     

       96
       +
       		return "", fmt.Errorf("failed to set algorithm: %w", setErr)

     

       97
        
       	}

     

       98
       +
       	if setErr := headers.Set(jws.TypeKey, "dpop+jwt"); setErr != nil {

     

       99
       +
       		return "", fmt.Errorf("failed to set type: %w", setErr)

     

       100
        
       	}

     

       101
        
       	// Set the public JWK directly - jwx library will handle serialization

     

       102
       +
       	if setErr := headers.Set(jws.JWKKey, pubKey); setErr != nil {

     

       103
       +
       		return "", fmt.Errorf("failed to set JWK: %w", setErr)

     

       104
        
       	}

     

       105
        
       

     

       106
        
       	// Sign using jws.Sign to preserve custom headers

+19 -9

internal/atproto/oauth/dpop_test.go

···

       28
        
       	}

     

       29
        
       

     

       30
        
       	// Decode and inspect the header

     

       31
       -
       	headerJSON, err := base64.RawURLEncoding.DecodeString(parts[0])

     

       32
       -
       	if err != nil {

     

       33
       -
       		t.Fatalf("Failed to decode header: %v", err)

     

       34
        
       	}

     

       35
        
       

     

       36
        
       	var header map[string]interface{}

     

       37
       -
       	if err := json.Unmarshal(headerJSON, &header); err != nil {

     

       38
       -
       		t.Fatalf("Failed to unmarshal header: %v", err)

     

       39
        
       	}

     

       40
        
       

     

       41
        
       	t.Logf("DPoP Header: %s", string(headerJSON))

     
···

       122
        
       

     

       123
        
       	// Decode payload

     

       124
        
       	parts := strings.Split(proof, ".")

     

       125
       -
       	payloadJSON, _ := base64.RawURLEncoding.DecodeString(parts[1])

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       126
        
       	var payload map[string]interface{}

     

       127
       -
       	json.Unmarshal(payloadJSON, &payload)

     

       0
        
       
     

       0
        
       
     

       128
        
       

     

       129
        
       	if payload["nonce"] != testNonce {

     

       130
        
       		t.Errorf("Expected nonce=%s, got %v", testNonce, payload["nonce"])

     
···

       146
        
       

     

       147
        
       	// Decode payload

     

       148
        
       	parts := strings.Split(proof, ".")

     

       149
       -
       	payloadJSON, _ := base64.RawURLEncoding.DecodeString(parts[1])

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       150
        
       	var payload map[string]interface{}

     

       151
       -
       	json.Unmarshal(payloadJSON, &payload)

     

       0
        
       
     

       0
        
       
     

       152
        
       

     

       153
        
       	ath, hasATH := payload["ath"]

     

       154
        
       	if !hasATH {

···

       28
        
       	}

     

       29
        
       

     

       30
        
       	// Decode and inspect the header

     

       31
       +
       	headerJSON, decodeErr := base64.RawURLEncoding.DecodeString(parts[0])

     

       32
       +
       	if decodeErr != nil {

     

       33
       +
       		t.Fatalf("Failed to decode header: %v", decodeErr)

     

       34
        
       	}

     

       35
        
       

     

       36
        
       	var header map[string]interface{}

     

       37
       +
       	if unmarshalErr := json.Unmarshal(headerJSON, &header); unmarshalErr != nil {

     

       38
       +
       		t.Fatalf("Failed to unmarshal header: %v", unmarshalErr)

     

       39
        
       	}

     

       40
        
       

     

       41
        
       	t.Logf("DPoP Header: %s", string(headerJSON))

     
···

       122
        
       

     

       123
        
       	// Decode payload

     

       124
        
       	parts := strings.Split(proof, ".")

     

       125
       +
       	payloadJSON, err := base64.RawURLEncoding.DecodeString(parts[1])

     

       126
       +
       	if err != nil {

     

       127
       +
       		t.Fatalf("Failed to decode payload: %v", err)

     

       128
       +
       	}

     

       129
        
       	var payload map[string]interface{}

     

       130
       +
       	if err := json.Unmarshal(payloadJSON, &payload); err != nil {

     

       131
       +
       		t.Fatalf("Failed to unmarshal payload: %v", err)

     

       132
       +
       	}

     

       133
        
       

     

       134
        
       	if payload["nonce"] != testNonce {

     

       135
        
       		t.Errorf("Expected nonce=%s, got %v", testNonce, payload["nonce"])

     
···

       151
        
       

     

       152
        
       	// Decode payload

     

       153
        
       	parts := strings.Split(proof, ".")

     

       154
       +
       	payloadJSON, err := base64.RawURLEncoding.DecodeString(parts[1])

     

       155
       +
       	if err != nil {

     

       156
       +
       		t.Fatalf("Failed to decode payload: %v", err)

     

       157
       +
       	}

     

       158
        
       	var payload map[string]interface{}

     

       159
       +
       	if err := json.Unmarshal(payloadJSON, &payload); err != nil {

     

       160
       +
       		t.Fatalf("Failed to unmarshal payload: %v", err)

     

       161
       +
       	}

     

       162
        
       

     

       163
        
       	ath, hasATH := payload["ath"]

     

       164
        
       	if !hasATH {

+16 -9

internal/atproto/xrpc/dpop_transport.go

···

       1
        
       package xrpc

     

       2
        
       

     

       3
        
       import (

     

       0
        
       
     

       4
        
       	"fmt"

     

       0
        
       
     

       5
        
       	"net/http"

     

       6
        
       	"sync"

     

       7
        
       

     

       8
       -
       	"Coves/internal/atproto/oauth"

     

       9
        
       	oauthCore "Coves/internal/core/oauth"

     

       10
        
       

     

       11
        
       	"github.com/lestrrat-go/jwx/v2/jwk"

     
···

       18
        
       // 3. Handles nonce rotation (retries on 401 with new nonce)

     

       19
        
       // 4. Updates nonces in session store

     

       20
        
       type DPoPTransport struct {

     

       21
       -
       	base            http.RoundTripper      // Underlying transport (usually http.DefaultTransport)

     

       22
       -
       	session         *oauthCore.OAuthSession // User's OAuth session

     

       23
       -
       	sessionStore    oauthCore.SessionStore  // For updating nonces

     

       24
       -
       	dpopKey         jwk.Key                 // Parsed DPoP private key

     

       25
       -
       	mu              sync.Mutex              // Protects nonce updates

     

       26
        
       }

     

       27
        
       

     

       28
        
       // NewDPoPTransport creates a new DPoP-enabled HTTP transport

     
···

       87
        
       			t.updateDPoPNonce(req.URL.String(), newNonce)

     

       88
        
       

     

       89
        
       			// Close the 401 response body

     

       90
       -
       			_ = resp.Body.Close()

     

       0
        
       
     

       0
        
       
     

       91
        
       

     

       92
        
       			// Retry with new nonce

     

       93
        
       			return t.retryWithNewNonce(req, newNonce)

     
···

       134
        
       		t.mu.Unlock()

     

       135
        
       		// Persist to database (async, best-effort)

     

       136
        
       		go func() {

     

       137
       -
       			_ = t.sessionStore.UpdatePDSNonce(did, newNonce)

     

       0
        
       
     

       0
        
       
     

       138
        
       		}()

     

       139
        
       		return

     

       140
        
       	}

     
···

       145
        
       		t.mu.Unlock()

     

       146
        
       		// Persist to database (async, best-effort)

     

       147
        
       		go func() {

     

       148
       -
       			_ = t.sessionStore.UpdateAuthServerNonce(did, newNonce)

     

       0
        
       
     

       0
        
       
     

       149
        
       		}()

     

       150
        
       		return

     

       151
        
       	}

···

       1
        
       package xrpc

     

       2
        
       

     

       3
        
       import (

     

       4
       +
       	"Coves/internal/atproto/oauth"

     

       5
        
       	"fmt"

     

       6
       +
       	"log"

     

       7
        
       	"net/http"

     

       8
        
       	"sync"

     

       9
        
       

     

       0
        
       
     

       10
        
       	oauthCore "Coves/internal/core/oauth"

     

       11
        
       

     

       12
        
       	"github.com/lestrrat-go/jwx/v2/jwk"

     
···

       19
        
       // 3. Handles nonce rotation (retries on 401 with new nonce)

     

       20
        
       // 4. Updates nonces in session store

     

       21
        
       type DPoPTransport struct {

     

       22
       +
       	base         http.RoundTripper       // Underlying transport (usually http.DefaultTransport)

     

       23
       +
       	session      *oauthCore.OAuthSession // User's OAuth session

     

       24
       +
       	sessionStore oauthCore.SessionStore  // For updating nonces

     

       25
       +
       	dpopKey      jwk.Key                 // Parsed DPoP private key

     

       26
       +
       	mu           sync.Mutex              // Protects nonce updates

     

       27
        
       }

     

       28
        
       

     

       29
        
       // NewDPoPTransport creates a new DPoP-enabled HTTP transport

     
···

       88
        
       			t.updateDPoPNonce(req.URL.String(), newNonce)

     

       89
        
       

     

       90
        
       			// Close the 401 response body

     

       91
       +
       			if err := resp.Body.Close(); err != nil {

     

       92
       +
       				log.Printf("Failed to close response body: %v", err)

     

       93
       +
       			}

     

       94
        
       

     

       95
        
       			// Retry with new nonce

     

       96
        
       			return t.retryWithNewNonce(req, newNonce)

     
···

       137
        
       		t.mu.Unlock()

     

       138
        
       		// Persist to database (async, best-effort)

     

       139
        
       		go func() {

     

       140
       +
       			if err := t.sessionStore.UpdatePDSNonce(did, newNonce); err != nil {

     

       141
       +
       				log.Printf("Failed to update PDS nonce: %v", err)

     

       142
       +
       			}

     

       143
        
       		}()

     

       144
        
       		return

     

       145
        
       	}

     
···

       150
        
       		t.mu.Unlock()

     

       151
        
       		// Persist to database (async, best-effort)

     

       152
        
       		go func() {

     

       153
       +
       			if err := t.sessionStore.UpdateAuthServerNonce(did, newNonce); err != nil {

     

       154
       +
       				log.Printf("Failed to update auth server nonce: %v", err)

     

       155
       +
       			}

     

       156
        
       		}()

     

       157
        
       		return

     

       158
        
       	}

+15 -46

internal/core/communities/service.go

···

       1
        
       package communities

     

       2
        
       

     

       3
        
       import (

     

       0
        
       
     

       4
        
       	"bytes"

     

       5
        
       	"context"

     

       6
        
       	"encoding/json"

     

       7
        
       	"fmt"

     

       8
        
       	"io"

     

       0
        
       
     

       9
        
       	"net/http"

     

       10
        
       	"regexp"

     

       11
        
       	"strings"

     

       12
        
       	"time"

     

       13
       -
       

     

       14
       -
       	"Coves/internal/atproto/did"

     

       15
        
       )

     

       16
        
       

     

       17
        
       // Community handle validation regex (DNS-valid handle: name.communities.instance.com)

     
···

       21
        
       type communityService struct {

     

       22
        
       	repo           Repository

     

       23
        
       	didGen         *did.Generator

     

       24
       -
       	pdsURL         string                 // PDS URL for write-forward operations

     

       25
       -
       	instanceDID    string                 // DID of this Coves instance

     

       26
       -
       	instanceDomain string                 // Domain of this instance (for handles)

     

       27
       -
       	pdsAccessToken string                 // Access token for authenticating to PDS as the instance

     

       28
       -
       	provisioner    *PDSAccountProvisioner // V2: Creates PDS accounts for communities

     

       29
        
       }

     

       30
        
       

     

       31
        
       // NewCommunityService creates a new community service

     

       32
       -
       func NewCommunityService(repo Repository, didGen *did.Generator, pdsURL string, instanceDID string, instanceDomain string, provisioner *PDSAccountProvisioner) Service {

     

       33
        
       	return &communityService{

     

       34
        
       		repo:           repo,

     

       35
        
       		didGen:         didGen,

     
···

       81
        
       	}

     

       82
        
       

     

       83
        
       	// Validate the atProto handle

     

       84
       -
       	if err := s.ValidateHandle(pdsAccount.Handle); err != nil {

     

       85
       -
       		return nil, fmt.Errorf("generated atProto handle is invalid: %w", err)

     

       86
        
       	}

     

       87
        
       

     

       88
        
       	// Build community profile record

     
···

       580
        
       	return s.callPDSWithAuth(ctx, "POST", endpoint, payload, accessToken)

     

       581
        
       }

     

       582
        
       

     

       583
       -
       func (s *communityService) putRecordOnPDS(ctx context.Context, repoDID, collection, rkey string, record map[string]interface{}) (string, string, error) {

     

       584
       -
       	endpoint := fmt.Sprintf("%s/xrpc/com.atproto.repo.putRecord", strings.TrimSuffix(s.pdsURL, "/"))

     

       585
       -
       

     

       586
       -
       	payload := map[string]interface{}{

     

       587
       -
       		"repo":       repoDID,

     

       588
       -
       		"collection": collection,

     

       589
       -
       		"rkey":       rkey,

     

       590
       -
       		"record":     record,

     

       591
       -
       	}

     

       592
       -
       

     

       593
       -
       	return s.callPDS(ctx, "POST", endpoint, payload)

     

       594
       -
       }

     

       595
       -
       

     

       596
        
       // putRecordOnPDSAs updates a record with a specific access token (for V2 community auth)

     

       597
        
       func (s *communityService) putRecordOnPDSAs(ctx context.Context, repoDID, collection, rkey string, record map[string]interface{}, accessToken string) (string, string, error) {

     

       598
        
       	endpoint := fmt.Sprintf("%s/xrpc/com.atproto.repo.putRecord", strings.TrimSuffix(s.pdsURL, "/"))

     
···

       660
        
       	if err != nil {

     

       661
        
       		return "", "", fmt.Errorf("failed to call PDS: %w", err)

     

       662
        
       	}

     

       663
       -
       	defer resp.Body.Close()

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       664
        
       

     

       665
        
       	body, err := io.ReadAll(resp.Body)

     

       666
        
       	if err != nil {

     
···

       688
        
       }

     

       689
        
       

     

       690
        
       // Helper functions

     

       691
       -
       

     

       692
       -
       func extractDomain(didOrURL string) string {

     

       693
       -
       	// For did:web:example.com -> example.com

     

       694
       -
       	if strings.HasPrefix(didOrURL, "did:web:") {

     

       695
       -
       		parts := strings.Split(didOrURL, ":")

     

       696
       -
       		if len(parts) >= 3 {

     

       697
       -
       			return parts[2]

     

       698
       -
       		}

     

       699
       -
       	}

     

       700
       -
       

     

       701
       -
       	// For URLs, extract domain

     

       702
       -
       	if strings.Contains(didOrURL, "://") {

     

       703
       -
       		parts := strings.Split(didOrURL, "://")

     

       704
       -
       		if len(parts) >= 2 {

     

       705
       -
       			domain := strings.Split(parts[1], "/")[0]

     

       706
       -
       			domain = strings.Split(domain, ":")[0] // Remove port

     

       707
       -
       			return domain

     

       708
       -
       		}

     

       709
       -
       	}

     

       710
       -
       

     

       711
       -
       	return ""

     

       712
       -
       }

     

       713
        
       

     

       714
        
       func extractRKeyFromURI(uri string) string {

     

       715
        
       	// at://did/collection/rkey -> rkey

···

       1
        
       package communities

     

       2
        
       

     

       3
        
       import (

     

       4
       +
       	"Coves/internal/atproto/did"

     

       5
        
       	"bytes"

     

       6
        
       	"context"

     

       7
        
       	"encoding/json"

     

       8
        
       	"fmt"

     

       9
        
       	"io"

     

       10
       +
       	"log"

     

       11
        
       	"net/http"

     

       12
        
       	"regexp"

     

       13
        
       	"strings"

     

       14
        
       	"time"

     

       0
        
       
     

       0
        
       
     

       15
        
       )

     

       16
        
       

     

       17
        
       // Community handle validation regex (DNS-valid handle: name.communities.instance.com)

     
···

       21
        
       type communityService struct {

     

       22
        
       	repo           Repository

     

       23
        
       	didGen         *did.Generator

     

       24
       +
       	provisioner    *PDSAccountProvisioner

     

       25
       +
       	pdsURL         string

     

       26
       +
       	instanceDID    string

     

       27
       +
       	instanceDomain string

     

       28
       +
       	pdsAccessToken string

     

       29
        
       }

     

       30
        
       

     

       31
        
       // NewCommunityService creates a new community service

     

       32
       +
       func NewCommunityService(repo Repository, didGen *did.Generator, pdsURL, instanceDID, instanceDomain string, provisioner *PDSAccountProvisioner) Service {

     

       33
        
       	return &communityService{

     

       34
        
       		repo:           repo,

     

       35
        
       		didGen:         didGen,

     
···

       81
        
       	}

     

       82
        
       

     

       83
        
       	// Validate the atProto handle

     

       84
       +
       	if validateErr := s.ValidateHandle(pdsAccount.Handle); validateErr != nil {

     

       85
       +
       		return nil, fmt.Errorf("generated atProto handle is invalid: %w", validateErr)

     

       86
        
       	}

     

       87
        
       

     

       88
        
       	// Build community profile record

     
···

       580
        
       	return s.callPDSWithAuth(ctx, "POST", endpoint, payload, accessToken)

     

       581
        
       }

     

       582
        
       

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       583
        
       // putRecordOnPDSAs updates a record with a specific access token (for V2 community auth)

     

       584
        
       func (s *communityService) putRecordOnPDSAs(ctx context.Context, repoDID, collection, rkey string, record map[string]interface{}, accessToken string) (string, string, error) {

     

       585
        
       	endpoint := fmt.Sprintf("%s/xrpc/com.atproto.repo.putRecord", strings.TrimSuffix(s.pdsURL, "/"))

     
···

       647
        
       	if err != nil {

     

       648
        
       		return "", "", fmt.Errorf("failed to call PDS: %w", err)

     

       649
        
       	}

     

       650
       +
       	defer func() {

     

       651
       +
       		if closeErr := resp.Body.Close(); closeErr != nil {

     

       652
       +
       			log.Printf("Failed to close response body: %v", closeErr)

     

       653
       +
       		}

     

       654
       +
       	}()

     

       655
        
       

     

       656
        
       	body, err := io.ReadAll(resp.Body)

     

       657
        
       	if err != nil {

     
···

       679
        
       }

     

       680
        
       

     

       681
        
       // Helper functions

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       682
        
       

     

       683
        
       func extractRKeyFromURI(uri string) string {

     

       684
        
       	// at://did/collection/rkey -> rkey

+4 -3

internal/core/oauth/auth_service.go

···

       1
        
       package oauth

     

       2
        
       

     

       3
        
       import (

     

       0
        
       
     

       4
        
       	"context"

     

       5
        
       	"fmt"

     

       6
        
       	"time"

     

       7
       -
       

     

       8
       -
       	"Coves/internal/atproto/oauth"

     

       9
        
       

     

       10
        
       	"github.com/lestrrat-go/jwx/v2/jwk"

     

       11
        
       )

     
···

       71
        
       	// Update nonce if provided (best effort - non-critical)

     

       72
        
       	if tokenResp.DpopAuthserverNonce != "" {

     

       73
        
       		session.DPoPAuthServerNonce = tokenResp.DpopAuthserverNonce

     

       74
       -
       		if err := s.sessionStore.UpdateAuthServerNonce(session.DID, tokenResp.DpopAuthserverNonce); err != nil {

     

       75
        
       			// Log but don't fail - nonce will be updated on next request

     

       0
        
       
     

       0
        
       
     

       76
        
       		}

     

       77
        
       	}

     

       78

···

       1
        
       package oauth

     

       2
        
       

     

       3
        
       import (

     

       4
       +
       	"Coves/internal/atproto/oauth"

     

       5
        
       	"context"

     

       6
        
       	"fmt"

     

       7
        
       	"time"

     

       0
        
       
     

       0
        
       
     

       8
        
       

     

       9
        
       	"github.com/lestrrat-go/jwx/v2/jwk"

     

       10
        
       )

     
···

       70
        
       	// Update nonce if provided (best effort - non-critical)

     

       71
        
       	if tokenResp.DpopAuthserverNonce != "" {

     

       72
        
       		session.DPoPAuthServerNonce = tokenResp.DpopAuthserverNonce

     

       73
       +
       		if updateErr := s.sessionStore.UpdateAuthServerNonce(session.DID, tokenResp.DpopAuthserverNonce); updateErr != nil {

     

       74
        
       			// Log but don't fail - nonce will be updated on next request

     

       75
       +
       			// (We ignore the error here intentionally as nonce updates are non-critical)

     

       76
       +
       			_ = updateErr

     

       77
        
       		}

     

       78
        
       	}

     

       79

+8 -4

internal/core/users/service.go

···

       1
        
       package users

     

       2
        
       

     

       3
        
       import (

     

       0
        
       
     

       4
        
       	"bytes"

     

       5
        
       	"context"

     

       6
        
       	"encoding/json"

     

       7
        
       	"fmt"

     

       8
        
       	"io"

     

       0
        
       
     

       9
        
       	"net/http"

     

       10
        
       	"regexp"

     

       11
        
       	"strings"

     

       12
        
       	"time"

     

       13
       -
       

     

       14
       -
       	"Coves/internal/atproto/identity"

     

       15
        
       )

     

       16
        
       

     

       17
        
       // atProto handle validation regex (per official atProto spec: https://atproto.com/specs/handle)

     
···

       187
        
       	if err != nil {

     

       188
        
       		return nil, fmt.Errorf("failed to call PDS: %w", err)

     

       189
        
       	}

     

       190
       -
       	defer resp.Body.Close()

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       191
        
       

     

       192
        
       	body, err := io.ReadAll(resp.Body)

     

       193
        
       	if err != nil {

     
···

       293
        
       	}

     

       294
        
       

     

       295
        
       	return nil

     

       296
       -
       }

···

       1
        
       package users

     

       2
        
       

     

       3
        
       import (

     

       4
       +
       	"Coves/internal/atproto/identity"

     

       5
        
       	"bytes"

     

       6
        
       	"context"

     

       7
        
       	"encoding/json"

     

       8
        
       	"fmt"

     

       9
        
       	"io"

     

       10
       +
       	"log"

     

       11
        
       	"net/http"

     

       12
        
       	"regexp"

     

       13
        
       	"strings"

     

       14
        
       	"time"

     

       0
        
       
     

       0
        
       
     

       15
        
       )

     

       16
        
       

     

       17
        
       // atProto handle validation regex (per official atProto spec: https://atproto.com/specs/handle)

     
···

       187
        
       	if err != nil {

     

       188
        
       		return nil, fmt.Errorf("failed to call PDS: %w", err)

     

       189
        
       	}

     

       190
       +
       	defer func() {

     

       191
       +
       		if closeErr := resp.Body.Close(); closeErr != nil {

     

       192
       +
       			log.Printf("Failed to close response body: %v", closeErr)

     

       193
       +
       		}

     

       194
       +
       	}()

     

       195
        
       

     

       196
        
       	body, err := io.ReadAll(resp.Body)

     

       197
        
       	if err != nil {

     
···

       297
        
       	}

     

       298
        
       

     

       299
        
       	return nil

     

       300
       +
       }

+20 -19

internal/db/postgres/community_repo.go

···

       1
        
       package postgres

     

       2
        
       

     

       3
        
       import (

     

       0
        
       
     

       4
        
       	"context"

     

       5
        
       	"database/sql"

     

       6
        
       	"fmt"

     

       0
        
       
     

       7
        
       	"strings"

     

       8
        
       

     

       9
       -
       	"Coves/internal/core/communities"

     

       10
        
       	"github.com/lib/pq"

     

       11
        
       )

     

       12
        
       

     
···

       44
        
       

     

       45
        
       	// Handle JSONB field - use sql.NullString with valid JSON or NULL

     

       46
        
       	var descFacets interface{}

     

       47
       -
       	if community.DescriptionFacets != nil && len(community.DescriptionFacets) > 0 {

     

       48
        
       		descFacets = community.DescriptionFacets

     

       49
        
       	} else {

     

       50
        
       		descFacets = nil

     
···

       82
        
       		nullString(community.RecordURI),

     

       83
        
       		nullString(community.RecordCID),

     

       84
        
       	).Scan(&community.ID, &community.CreatedAt, &community.UpdatedAt)

     

       85
       -
       

     

       86
        
       	if err != nil {

     

       87
        
       		// Check for unique constraint violations

     

       88
        
       		if strings.Contains(err.Error(), "duplicate key") {

     
···

       240
        
       

     

       241
        
       	// Handle JSONB field - use sql.NullString with valid JSON or NULL

     

       242
        
       	var descFacets interface{}

     

       243
       -
       	if community.DescriptionFacets != nil && len(community.DescriptionFacets) > 0 {

     

       244
        
       		descFacets = community.DescriptionFacets

     

       245
        
       	} else {

     

       246
        
       		descFacets = nil

     
···

       358
        
       	if err != nil {

     

       359
        
       		return nil, 0, fmt.Errorf("failed to list communities: %w", err)

     

       360
        
       	}

     

       361
       -
       	defer rows.Close()

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       362
        
       

     

       363
        
       	result := []*communities.Community{}

     

       364
        
       	for rows.Next() {

     
···

       368
        
       		var descFacets []byte

     

       369
        
       		var contentWarnings []string

     

       370
        
       

     

       371
       -
       		err := rows.Scan(

     

       372
        
       			&community.ID, &community.DID, &community.Handle, &community.Name,

     

       373
        
       			&displayName, &description, &descFacets,

     

       374
        
       			&avatarCID, &bannerCID,

     
···

       380
        
       			&community.CreatedAt, &community.UpdatedAt,

     

       381
        
       			&recordURI, &recordCID,

     

       382
        
       		)

     

       383
       -
       		if err != nil {

     

       384
       -
       			return nil, 0, fmt.Errorf("failed to scan community: %w", err)

     

       385
        
       		}

     

       386
        
       

     

       387
        
       		// Map nullable fields

     
···

       456
        
       	if err != nil {

     

       457
        
       		return nil, 0, fmt.Errorf("failed to search communities: %w", err)

     

       458
        
       	}

     

       459
       -
       	defer rows.Close()

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       460
        
       

     

       461
        
       	result := []*communities.Community{}

     

       462
        
       	for rows.Next() {

     
···

       467
        
       		var contentWarnings []string

     

       468
        
       		var relevance float64

     

       469
        
       

     

       470
       -
       		err := rows.Scan(

     

       471
        
       			&community.ID, &community.DID, &community.Handle, &community.Name,

     

       472
        
       			&displayName, &description, &descFacets,

     

       473
        
       			&avatarCID, &bannerCID,

     
···

       480
        
       			&recordURI, &recordCID,

     

       481
        
       			&relevance,

     

       482
        
       		)

     

       483
       -
       		if err != nil {

     

       484
       -
       			return nil, 0, fmt.Errorf("failed to scan community: %w", err)

     

       485
        
       		}

     

       486
        
       

     

       487
        
       		// Map nullable fields

     
···

       513
        
       func nullString(s string) sql.NullString {

     

       514
        
       	return sql.NullString{String: s, Valid: s != ""}

     

       515
        
       }

     

       516
       -
       

     

       517
       -
       func nullBytes(b []byte) []byte {

     

       518
       -
       	if b == nil || len(b) == 0 {

     

       519
       -
       		return nil

     

       520
       -
       	}

     

       521
       -
       	return b

     

       522
       -
       }

···

       1
        
       package postgres

     

       2
        
       

     

       3
        
       import (

     

       4
       +
       	"Coves/internal/core/communities"

     

       5
        
       	"context"

     

       6
        
       	"database/sql"

     

       7
        
       	"fmt"

     

       8
       +
       	"log"

     

       9
        
       	"strings"

     

       10
        
       

     

       0
        
       
     

       11
        
       	"github.com/lib/pq"

     

       12
        
       )

     

       13
        
       

     
···

       45
        
       

     

       46
        
       	// Handle JSONB field - use sql.NullString with valid JSON or NULL

     

       47
        
       	var descFacets interface{}

     

       48
       +
       	if len(community.DescriptionFacets) > 0 {

     

       49
        
       		descFacets = community.DescriptionFacets

     

       50
        
       	} else {

     

       51
        
       		descFacets = nil

     
···

       83
        
       		nullString(community.RecordURI),

     

       84
        
       		nullString(community.RecordCID),

     

       85
        
       	).Scan(&community.ID, &community.CreatedAt, &community.UpdatedAt)

     

       0
        
       
     

       86
        
       	if err != nil {

     

       87
        
       		// Check for unique constraint violations

     

       88
        
       		if strings.Contains(err.Error(), "duplicate key") {

     
···

       240
        
       

     

       241
        
       	// Handle JSONB field - use sql.NullString with valid JSON or NULL

     

       242
        
       	var descFacets interface{}

     

       243
       +
       	if len(community.DescriptionFacets) > 0 {

     

       244
        
       		descFacets = community.DescriptionFacets

     

       245
        
       	} else {

     

       246
        
       		descFacets = nil

     
···

       358
        
       	if err != nil {

     

       359
        
       		return nil, 0, fmt.Errorf("failed to list communities: %w", err)

     

       360
        
       	}

     

       361
       +
       	defer func() {

     

       362
       +
       		if closeErr := rows.Close(); closeErr != nil {

     

       363
       +
       			log.Printf("Failed to close rows: %v", closeErr)

     

       364
       +
       		}

     

       365
       +
       	}()

     

       366
        
       

     

       367
        
       	result := []*communities.Community{}

     

       368
        
       	for rows.Next() {

     
···

       372
        
       		var descFacets []byte

     

       373
        
       		var contentWarnings []string

     

       374
        
       

     

       375
       +
       		scanErr := rows.Scan(

     

       376
        
       			&community.ID, &community.DID, &community.Handle, &community.Name,

     

       377
        
       			&displayName, &description, &descFacets,

     

       378
        
       			&avatarCID, &bannerCID,

     
···

       384
        
       			&community.CreatedAt, &community.UpdatedAt,

     

       385
        
       			&recordURI, &recordCID,

     

       386
        
       		)

     

       387
       +
       		if scanErr != nil {

     

       388
       +
       			return nil, 0, fmt.Errorf("failed to scan community: %w", scanErr)

     

       389
        
       		}

     

       390
        
       

     

       391
        
       		// Map nullable fields

     
···

       460
        
       	if err != nil {

     

       461
        
       		return nil, 0, fmt.Errorf("failed to search communities: %w", err)

     

       462
        
       	}

     

       463
       +
       	defer func() {

     

       464
       +
       		if closeErr := rows.Close(); closeErr != nil {

     

       465
       +
       			log.Printf("Failed to close rows: %v", closeErr)

     

       466
       +
       		}

     

       467
       +
       	}()

     

       468
        
       

     

       469
        
       	result := []*communities.Community{}

     

       470
        
       	for rows.Next() {

     
···

       475
        
       		var contentWarnings []string

     

       476
        
       		var relevance float64

     

       477
        
       

     

       478
       +
       		scanErr := rows.Scan(

     

       479
        
       			&community.ID, &community.DID, &community.Handle, &community.Name,

     

       480
        
       			&displayName, &description, &descFacets,

     

       481
        
       			&avatarCID, &bannerCID,

     
···

       488
        
       			&recordURI, &recordCID,

     

       489
        
       			&relevance,

     

       490
        
       		)

     

       491
       +
       		if scanErr != nil {

     

       492
       +
       			return nil, 0, fmt.Errorf("failed to scan community: %w", scanErr)

     

       493
        
       		}

     

       494
        
       

     

       495
        
       		// Map nullable fields

     
···

       521
        
       func nullString(s string) sql.NullString {

     

       522
        
       	return sql.NullString{String: s, Valid: s != ""}

     

       523
        
       }

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0

+18 -12

internal/db/postgres/community_repo_memberships.go

···

       1
        
       package postgres

     

       2
        
       

     

       3
        
       import (

     

       0
        
       
     

       4
        
       	"context"

     

       5
        
       	"database/sql"

     

       6
        
       	"fmt"

     

       0
        
       
     

       7
        
       	"strings"

     

       8
       -
       

     

       9
       -
       	"Coves/internal/core/communities"

     

       10
        
       )

     

       11
        
       

     

       12
        
       // CreateMembership creates a new membership record

     
···

       29
        
       		membership.IsBanned,

     

       30
        
       		membership.IsModerator,

     

       31
        
       	).Scan(&membership.ID, &membership.JoinedAt, &membership.LastActiveAt)

     

       32
       -
       

     

       33
        
       	if err != nil {

     

       34
        
       		if strings.Contains(err.Error(), "duplicate key") {

     

       35
        
       			return nil, fmt.Errorf("membership already exists")

     
···

       120
        
       	if err != nil {

     

       121
        
       		return nil, fmt.Errorf("failed to list members: %w", err)

     

       122
        
       	}

     

       123
       -
       	defer rows.Close()

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       124
        
       

     

       125
        
       	result := []*communities.Membership{}

     

       126
        
       	for rows.Next() {

     

       127
        
       		membership := &communities.Membership{}

     

       128
        
       

     

       129
       -
       		err := rows.Scan(

     

       130
        
       			&membership.ID,

     

       131
        
       			&membership.UserDID,

     

       132
        
       			&membership.CommunityDID,

     
···

       137
        
       			&membership.IsBanned,

     

       138
        
       			&membership.IsModerator,

     

       139
        
       		)

     

       140
       -
       		if err != nil {

     

       141
       -
       			return nil, fmt.Errorf("failed to scan member: %w", err)

     

       142
        
       		}

     

       143
        
       

     

       144
        
       		result = append(result, membership)

     
···

       169
        
       		action.CreatedAt,

     

       170
        
       		action.ExpiresAt,

     

       171
        
       	).Scan(&action.ID, &action.CreatedAt)

     

       172
       -
       

     

       173
        
       	if err != nil {

     

       174
        
       		if strings.Contains(err.Error(), "foreign key") {

     

       175
        
       			return nil, communities.ErrCommunityNotFound

     
···

       193
        
       	if err != nil {

     

       194
        
       		return nil, fmt.Errorf("failed to list moderation actions: %w", err)

     

       195
        
       	}

     

       196
       -
       	defer rows.Close()

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       197
        
       

     

       198
        
       	result := []*communities.ModerationAction{}

     

       199
        
       	for rows.Next() {

     

       200
        
       		action := &communities.ModerationAction{}

     

       201
        
       		var reason sql.NullString

     

       202
        
       

     

       203
       -
       		err := rows.Scan(

     

       204
        
       			&action.ID,

     

       205
        
       			&action.CommunityDID,

     

       206
        
       			&action.Action,

     
···

       210
        
       			&action.CreatedAt,

     

       211
        
       			&action.ExpiresAt,

     

       212
        
       		)

     

       213
       -
       		if err != nil {

     

       214
       -
       			return nil, fmt.Errorf("failed to scan moderation action: %w", err)

     

       215
        
       		}

     

       216
        
       

     

       217
        
       		action.Reason = reason.String

···

       1
        
       package postgres

     

       2
        
       

     

       3
        
       import (

     

       4
       +
       	"Coves/internal/core/communities"

     

       5
        
       	"context"

     

       6
        
       	"database/sql"

     

       7
        
       	"fmt"

     

       8
       +
       	"log"

     

       9
        
       	"strings"

     

       0
        
       
     

       0
        
       
     

       10
        
       )

     

       11
        
       

     

       12
        
       // CreateMembership creates a new membership record

     
···

       29
        
       		membership.IsBanned,

     

       30
        
       		membership.IsModerator,

     

       31
        
       	).Scan(&membership.ID, &membership.JoinedAt, &membership.LastActiveAt)

     

       0
        
       
     

       32
        
       	if err != nil {

     

       33
        
       		if strings.Contains(err.Error(), "duplicate key") {

     

       34
        
       			return nil, fmt.Errorf("membership already exists")

     
···

       119
        
       	if err != nil {

     

       120
        
       		return nil, fmt.Errorf("failed to list members: %w", err)

     

       121
        
       	}

     

       122
       +
       	defer func() {

     

       123
       +
       		if closeErr := rows.Close(); closeErr != nil {

     

       124
       +
       			log.Printf("Failed to close rows: %v", closeErr)

     

       125
       +
       		}

     

       126
       +
       	}()

     

       127
        
       

     

       128
        
       	result := []*communities.Membership{}

     

       129
        
       	for rows.Next() {

     

       130
        
       		membership := &communities.Membership{}

     

       131
        
       

     

       132
       +
       		scanErr := rows.Scan(

     

       133
        
       			&membership.ID,

     

       134
        
       			&membership.UserDID,

     

       135
        
       			&membership.CommunityDID,

     
···

       140
        
       			&membership.IsBanned,

     

       141
        
       			&membership.IsModerator,

     

       142
        
       		)

     

       143
       +
       		if scanErr != nil {

     

       144
       +
       			return nil, fmt.Errorf("failed to scan member: %w", scanErr)

     

       145
        
       		}

     

       146
        
       

     

       147
        
       		result = append(result, membership)

     
···

       172
        
       		action.CreatedAt,

     

       173
        
       		action.ExpiresAt,

     

       174
        
       	).Scan(&action.ID, &action.CreatedAt)

     

       0
        
       
     

       175
        
       	if err != nil {

     

       176
        
       		if strings.Contains(err.Error(), "foreign key") {

     

       177
        
       			return nil, communities.ErrCommunityNotFound

     
···

       195
        
       	if err != nil {

     

       196
        
       		return nil, fmt.Errorf("failed to list moderation actions: %w", err)

     

       197
        
       	}

     

       198
       +
       	defer func() {

     

       199
       +
       		if closeErr := rows.Close(); closeErr != nil {

     

       200
       +
       			log.Printf("Failed to close rows: %v", closeErr)

     

       201
       +
       		}

     

       202
       +
       	}()

     

       203
        
       

     

       204
        
       	result := []*communities.ModerationAction{}

     

       205
        
       	for rows.Next() {

     

       206
        
       		action := &communities.ModerationAction{}

     

       207
        
       		var reason sql.NullString

     

       208
        
       

     

       209
       +
       		scanErr := rows.Scan(

     

       210
        
       			&action.ID,

     

       211
        
       			&action.CommunityDID,

     

       212
        
       			&action.Action,

     
···

       216
        
       			&action.CreatedAt,

     

       217
        
       			&action.ExpiresAt,

     

       218
        
       		)

     

       219
       +
       		if scanErr != nil {

     

       220
       +
       			return nil, fmt.Errorf("failed to scan moderation action: %w", scanErr)

     

       221
        
       		}

     

       222
        
       

     

       223
        
       		action.Reason = reason.String

+32 -17

internal/db/postgres/community_repo_subscriptions.go

···

       1
        
       package postgres

     

       2
        
       

     

       3
        
       import (

     

       0
        
       
     

       4
        
       	"context"

     

       5
        
       	"database/sql"

     

       6
        
       	"fmt"

     

       0
        
       
     

       7
        
       	"strings"

     

       8
       -
       

     

       9
       -
       	"Coves/internal/core/communities"

     

       10
        
       )

     

       11
        
       

     

       12
        
       // Subscribe creates a new subscription record

     
···

       23
        
       		nullString(subscription.RecordURI),

     

       24
        
       		nullString(subscription.RecordCID),

     

       25
        
       	).Scan(&subscription.ID, &subscription.SubscribedAt)

     

       26
       -
       

     

       27
        
       	if err != nil {

     

       28
        
       		if strings.Contains(err.Error(), "duplicate key") {

     

       29
        
       			return nil, communities.ErrSubscriptionAlreadyExists

     
···

       44
        
       	if err != nil {

     

       45
        
       		return nil, fmt.Errorf("failed to begin transaction: %w", err)

     

       46
        
       	}

     

       47
       -
       	defer tx.Rollback()

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       48
        
       

     

       49
        
       	// Insert subscription with ON CONFLICT DO NOTHING for idempotency

     

       50
        
       	query := `

     
···

       70
        
       			return nil, fmt.Errorf("failed to get existing subscription: %w", err)

     

       71
        
       		}

     

       72
        
       		// Don't increment count - subscription already existed

     

       73
       -
       		if err := tx.Commit(); err != nil {

     

       74
       -
       			return nil, fmt.Errorf("failed to commit transaction: %w", err)

     

       75
        
       		}

     

       76
        
       		return subscription, nil

     

       77
        
       	}

     
···

       129
        
       	if err != nil {

     

       130
        
       		return fmt.Errorf("failed to begin transaction: %w", err)

     

       131
        
       	}

     

       132
       -
       	defer tx.Rollback()

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       133
        
       

     

       134
        
       	// Delete subscription

     

       135
        
       	deleteQuery := `DELETE FROM community_subscriptions WHERE user_did = $1 AND community_did = $2`

     
···

       145
        
       

     

       146
        
       	// If no rows deleted, subscription didn't exist (idempotent - not an error)

     

       147
        
       	if rowsAffected == 0 {

     

       148
       -
       		if err := tx.Commit(); err != nil {

     

       149
       -
       			return fmt.Errorf("failed to commit transaction: %w", err)

     

       150
        
       		}

     

       151
        
       		return nil

     

       152
        
       	}

     
···

       214
        
       	if err != nil {

     

       215
        
       		return nil, fmt.Errorf("failed to list subscriptions: %w", err)

     

       216
        
       	}

     

       217
       -
       	defer rows.Close()

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       218
        
       

     

       219
        
       	result := []*communities.Subscription{}

     

       220
        
       	for rows.Next() {

     

       221
        
       		subscription := &communities.Subscription{}

     

       222
        
       		var recordURI, recordCID sql.NullString

     

       223
        
       

     

       224
       -
       		err := rows.Scan(

     

       225
        
       			&subscription.ID,

     

       226
        
       			&subscription.UserDID,

     

       227
        
       			&subscription.CommunityDID,

     
···

       229
        
       			&recordURI,

     

       230
        
       			&recordCID,

     

       231
        
       		)

     

       232
       -
       		if err != nil {

     

       233
       -
       			return nil, fmt.Errorf("failed to scan subscription: %w", err)

     

       234
        
       		}

     

       235
        
       

     

       236
        
       		subscription.RecordURI = recordURI.String

     
···

       259
        
       	if err != nil {

     

       260
        
       		return nil, fmt.Errorf("failed to list subscribers: %w", err)

     

       261
        
       	}

     

       262
       -
       	defer rows.Close()

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       263
        
       

     

       264
        
       	result := []*communities.Subscription{}

     

       265
        
       	for rows.Next() {

     

       266
        
       		subscription := &communities.Subscription{}

     

       267
        
       		var recordURI, recordCID sql.NullString

     

       268
        
       

     

       269
       -
       		err := rows.Scan(

     

       270
        
       			&subscription.ID,

     

       271
        
       			&subscription.UserDID,

     

       272
        
       			&subscription.CommunityDID,

     
···

       274
        
       			&recordURI,

     

       275
        
       			&recordCID,

     

       276
        
       		)

     

       277
       -
       		if err != nil {

     

       278
       -
       			return nil, fmt.Errorf("failed to scan subscriber: %w", err)

     

       279
        
       		}

     

       280
        
       

     

       281
        
       		subscription.RecordURI = recordURI.String

···

       1
        
       package postgres

     

       2
        
       

     

       3
        
       import (

     

       4
       +
       	"Coves/internal/core/communities"

     

       5
        
       	"context"

     

       6
        
       	"database/sql"

     

       7
        
       	"fmt"

     

       8
       +
       	"log"

     

       9
        
       	"strings"

     

       0
        
       
     

       0
        
       
     

       10
        
       )

     

       11
        
       

     

       12
        
       // Subscribe creates a new subscription record

     
···

       23
        
       		nullString(subscription.RecordURI),

     

       24
        
       		nullString(subscription.RecordCID),

     

       25
        
       	).Scan(&subscription.ID, &subscription.SubscribedAt)

     

       0
        
       
     

       26
        
       	if err != nil {

     

       27
        
       		if strings.Contains(err.Error(), "duplicate key") {

     

       28
        
       			return nil, communities.ErrSubscriptionAlreadyExists

     
···

       43
        
       	if err != nil {

     

       44
        
       		return nil, fmt.Errorf("failed to begin transaction: %w", err)

     

       45
        
       	}

     

       46
       +
       	defer func() {

     

       47
       +
       		if rollbackErr := tx.Rollback(); rollbackErr != nil && rollbackErr != sql.ErrTxDone {

     

       48
       +
       			log.Printf("Failed to rollback transaction: %v", rollbackErr)

     

       49
       +
       		}

     

       50
       +
       	}()

     

       51
        
       

     

       52
        
       	// Insert subscription with ON CONFLICT DO NOTHING for idempotency

     

       53
        
       	query := `

     
···

       73
        
       			return nil, fmt.Errorf("failed to get existing subscription: %w", err)

     

       74
        
       		}

     

       75
        
       		// Don't increment count - subscription already existed

     

       76
       +
       		if commitErr := tx.Commit(); commitErr != nil {

     

       77
       +
       			return nil, fmt.Errorf("failed to commit transaction: %w", commitErr)

     

       78
        
       		}

     

       79
        
       		return subscription, nil

     

       80
        
       	}

     
···

       132
        
       	if err != nil {

     

       133
        
       		return fmt.Errorf("failed to begin transaction: %w", err)

     

       134
        
       	}

     

       135
       +
       	defer func() {

     

       136
       +
       		if rollbackErr := tx.Rollback(); rollbackErr != nil && rollbackErr != sql.ErrTxDone {

     

       137
       +
       			log.Printf("Failed to rollback transaction: %v", rollbackErr)

     

       138
       +
       		}

     

       139
       +
       	}()

     

       140
        
       

     

       141
        
       	// Delete subscription

     

       142
        
       	deleteQuery := `DELETE FROM community_subscriptions WHERE user_did = $1 AND community_did = $2`

     
···

       152
        
       

     

       153
        
       	// If no rows deleted, subscription didn't exist (idempotent - not an error)

     

       154
        
       	if rowsAffected == 0 {

     

       155
       +
       		if commitErr := tx.Commit(); commitErr != nil {

     

       156
       +
       			return fmt.Errorf("failed to commit transaction: %w", commitErr)

     

       157
        
       		}

     

       158
        
       		return nil

     

       159
        
       	}

     
···

       221
        
       	if err != nil {

     

       222
        
       		return nil, fmt.Errorf("failed to list subscriptions: %w", err)

     

       223
        
       	}

     

       224
       +
       	defer func() {

     

       225
       +
       		if closeErr := rows.Close(); closeErr != nil {

     

       226
       +
       			log.Printf("Failed to close rows: %v", closeErr)

     

       227
       +
       		}

     

       228
       +
       	}()

     

       229
        
       

     

       230
        
       	result := []*communities.Subscription{}

     

       231
        
       	for rows.Next() {

     

       232
        
       		subscription := &communities.Subscription{}

     

       233
        
       		var recordURI, recordCID sql.NullString

     

       234
        
       

     

       235
       +
       		scanErr := rows.Scan(

     

       236
        
       			&subscription.ID,

     

       237
        
       			&subscription.UserDID,

     

       238
        
       			&subscription.CommunityDID,

     
···

       240
        
       			&recordURI,

     

       241
        
       			&recordCID,

     

       242
        
       		)

     

       243
       +
       		if scanErr != nil {

     

       244
       +
       			return nil, fmt.Errorf("failed to scan subscription: %w", scanErr)

     

       245
        
       		}

     

       246
        
       

     

       247
        
       		subscription.RecordURI = recordURI.String

     
···

       270
        
       	if err != nil {

     

       271
        
       		return nil, fmt.Errorf("failed to list subscribers: %w", err)

     

       272
        
       	}

     

       273
       +
       	defer func() {

     

       274
       +
       		if closeErr := rows.Close(); closeErr != nil {

     

       275
       +
       			log.Printf("Failed to close rows: %v", closeErr)

     

       276
       +
       		}

     

       277
       +
       	}()

     

       278
        
       

     

       279
        
       	result := []*communities.Subscription{}

     

       280
        
       	for rows.Next() {

     

       281
        
       		subscription := &communities.Subscription{}

     

       282
        
       		var recordURI, recordCID sql.NullString

     

       283
        
       

     

       284
       +
       		scanErr := rows.Scan(

     

       285
        
       			&subscription.ID,

     

       286
        
       			&subscription.UserDID,

     

       287
        
       			&subscription.CommunityDID,

     
···

       289
        
       			&recordURI,

     

       290
        
       			&recordCID,

     

       291
        
       		)

     

       292
       +
       		if scanErr != nil {

     

       293
       +
       			return nil, fmt.Errorf("failed to scan subscriber: %w", scanErr)

     

       294
        
       		}

     

       295
        
       

     

       296
        
       		subscription.RecordURI = recordURI.String