bretton.dev
A community based topic aggregation platform built on atproto
refactor(votes): remove vote write-forward XRPC endpoints
Remove all vote write endpoints from the AppView. Vote creation and
deletion are now performed directly by clients at their PDS using
standard atProto repo operations.
Removed:
- POST /xrpc/social.coves.interaction.createVote handler
- POST /xrpc/social.coves.interaction.deleteVote handler
- Vote route registration
- E2E tests for write-forward pattern (~800 lines)
Endpoints now return 404. Clients should use:
- com.atproto.repo.createRecord (collection: social.coves.interaction.vote)
- com.atproto.repo.deleteRecord (extract rkey from vote URI)
The AppView indexes votes from Jetstream for aggregation and querying.
Deleted files:
- internal/api/handlers/vote/create_vote.go
- internal/api/handlers/vote/delete_vote.go
- internal/api/routes/vote.go
- tests/integration/vote_e2e_test.go
-1017
-129
internal/api/handlers/vote/create_vote.go
-129
internal/api/handlers/vote/create_vote.go
···
1
-
package vote
2
-
3
-
import (
4
-
"Coves/internal/api/handlers"
5
-
"Coves/internal/api/middleware"
6
-
"Coves/internal/core/votes"
7
-
"encoding/json"
8
-
"log"
9
-
"net/http"
10
-
)
11
-
12
-
// CreateVoteHandler handles vote creation
13
-
type CreateVoteHandler struct {
14
-
service votes.Service
15
-
}
16
-
17
-
// NewCreateVoteHandler creates a new create vote handler
18
-
func NewCreateVoteHandler(service votes.Service) *CreateVoteHandler {
19
-
return &CreateVoteHandler{
20
-
service: service,
21
-
}
22
-
}
23
-
24
-
// HandleCreateVote creates a vote or toggles an existing vote
25
-
// POST /xrpc/social.coves.interaction.createVote
26
-
//
27
-
// Request body: { "subject": "at://...", "direction": "up" | "down" }
28
-
func (h *CreateVoteHandler) HandleCreateVote(w http.ResponseWriter, r *http.Request) {
29
-
if r.Method != http.MethodPost {
30
-
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
31
-
return
32
-
}
33
-
34
-
// Parse request body
35
-
var req votes.CreateVoteRequest
36
-
37
-
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
38
-
handlers.WriteError(w, http.StatusBadRequest, "InvalidRequest", "Invalid request body")
39
-
return
40
-
}
41
-
42
-
if req.Subject == "" {
43
-
handlers.WriteError(w, http.StatusBadRequest, "InvalidRequest", "subject is required")
44
-
return
45
-
}
46
-
47
-
if req.Direction == "" {
48
-
handlers.WriteError(w, http.StatusBadRequest, "InvalidRequest", "direction is required")
49
-
return
50
-
}
51
-
52
-
if req.Direction != "up" && req.Direction != "down" {
53
-
handlers.WriteError(w, http.StatusBadRequest, "InvalidRequest", "direction must be 'up' or 'down'")
54
-
return
55
-
}
56
-
57
-
// Extract authenticated user DID and access token from request context (injected by auth middleware)
58
-
voterDID := middleware.GetUserDID(r)
59
-
if voterDID == "" {
60
-
handlers.WriteError(w, http.StatusUnauthorized, "AuthRequired", "Authentication required")
61
-
return
62
-
}
63
-
64
-
userAccessToken := middleware.GetUserAccessToken(r)
65
-
if userAccessToken == "" {
66
-
handlers.WriteError(w, http.StatusUnauthorized, "AuthRequired", "Missing access token")
67
-
return
68
-
}
69
-
70
-
// Create vote via service (write-forward to user's PDS)
71
-
response, err := h.service.CreateVote(r.Context(), voterDID, userAccessToken, req)
72
-
if err != nil {
73
-
handleServiceError(w, err)
74
-
return
75
-
}
76
-
77
-
// Handle toggle-off case (vote was deleted, not created)
78
-
if response.URI == "" {
79
-
// Vote was toggled off (deleted)
80
-
w.Header().Set("Content-Type", "application/json")
81
-
w.WriteHeader(http.StatusOK)
82
-
if err := json.NewEncoder(w).Encode(map[string]interface{}{
83
-
"deleted": true,
84
-
}); err != nil {
85
-
log.Printf("Failed to encode response: %v", err)
86
-
}
87
-
return
88
-
}
89
-
90
-
// Return success response
91
-
responseMap := map[string]interface{}{
92
-
"uri": response.URI,
93
-
"cid": response.CID,
94
-
}
95
-
96
-
if response.Existing != nil {
97
-
responseMap["existing"] = *response.Existing
98
-
}
99
-
100
-
w.Header().Set("Content-Type", "application/json")
101
-
w.WriteHeader(http.StatusOK)
102
-
if err := json.NewEncoder(w).Encode(responseMap); err != nil {
103
-
log.Printf("Failed to encode response: %v", err)
104
-
}
105
-
}
106
-
107
-
// handleServiceError converts service errors to HTTP responses
108
-
func handleServiceError(w http.ResponseWriter, err error) {
109
-
switch err {
110
-
case votes.ErrVoteNotFound:
111
-
handlers.WriteError(w, http.StatusNotFound, "VoteNotFound", "Vote not found")
112
-
case votes.ErrSubjectNotFound:
113
-
handlers.WriteError(w, http.StatusNotFound, "SubjectNotFound", "Post or comment not found")
114
-
case votes.ErrInvalidDirection:
115
-
handlers.WriteError(w, http.StatusBadRequest, "InvalidRequest", "Invalid vote direction")
116
-
case votes.ErrInvalidSubject:
117
-
handlers.WriteError(w, http.StatusBadRequest, "InvalidRequest", "Invalid subject URI")
118
-
case votes.ErrVoteAlreadyExists:
119
-
handlers.WriteError(w, http.StatusConflict, "VoteAlreadyExists", "Vote already exists")
120
-
case votes.ErrNotAuthorized:
121
-
handlers.WriteError(w, http.StatusForbidden, "NotAuthorized", "Not authorized")
122
-
case votes.ErrBanned:
123
-
handlers.WriteError(w, http.StatusForbidden, "Banned", "User is banned from this community")
124
-
default:
125
-
// Check for validation errors
126
-
log.Printf("Vote creation error: %v", err)
127
-
handlers.WriteError(w, http.StatusInternalServerError, "InternalError", "Failed to create vote")
128
-
}
129
-
}
-75
internal/api/handlers/vote/delete_vote.go
-75
internal/api/handlers/vote/delete_vote.go
···
1
-
package vote
2
-
3
-
import (
4
-
"Coves/internal/api/handlers"
5
-
"Coves/internal/api/middleware"
6
-
"Coves/internal/core/votes"
7
-
"encoding/json"
8
-
"log"
9
-
"net/http"
10
-
)
11
-
12
-
// DeleteVoteHandler handles vote deletion
13
-
type DeleteVoteHandler struct {
14
-
service votes.Service
15
-
}
16
-
17
-
// NewDeleteVoteHandler creates a new delete vote handler
18
-
func NewDeleteVoteHandler(service votes.Service) *DeleteVoteHandler {
19
-
return &DeleteVoteHandler{
20
-
service: service,
21
-
}
22
-
}
23
-
24
-
// HandleDeleteVote removes a vote from a post/comment
25
-
// POST /xrpc/social.coves.interaction.deleteVote
26
-
//
27
-
// Request body: { "subject": "at://..." }
28
-
func (h *DeleteVoteHandler) HandleDeleteVote(w http.ResponseWriter, r *http.Request) {
29
-
if r.Method != http.MethodPost {
30
-
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
31
-
return
32
-
}
33
-
34
-
// Parse request body
35
-
var req votes.DeleteVoteRequest
36
-
37
-
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
38
-
handlers.WriteError(w, http.StatusBadRequest, "InvalidRequest", "Invalid request body")
39
-
return
40
-
}
41
-
42
-
if req.Subject == "" {
43
-
handlers.WriteError(w, http.StatusBadRequest, "InvalidRequest", "subject is required")
44
-
return
45
-
}
46
-
47
-
// Extract authenticated user DID and access token from request context (injected by auth middleware)
48
-
voterDID := middleware.GetUserDID(r)
49
-
if voterDID == "" {
50
-
handlers.WriteError(w, http.StatusUnauthorized, "AuthRequired", "Authentication required")
51
-
return
52
-
}
53
-
54
-
userAccessToken := middleware.GetUserAccessToken(r)
55
-
if userAccessToken == "" {
56
-
handlers.WriteError(w, http.StatusUnauthorized, "AuthRequired", "Missing access token")
57
-
return
58
-
}
59
-
60
-
// Delete vote via service (delete record on PDS)
61
-
err := h.service.DeleteVote(r.Context(), voterDID, userAccessToken, req)
62
-
if err != nil {
63
-
handleServiceError(w, err)
64
-
return
65
-
}
66
-
67
-
// Return success response
68
-
w.Header().Set("Content-Type", "application/json")
69
-
w.WriteHeader(http.StatusOK)
70
-
if err := json.NewEncoder(w).Encode(map[string]interface{}{
71
-
"success": true,
72
-
}); err != nil {
73
-
log.Printf("Failed to encode response: %v", err)
74
-
}
75
-
}
-24
internal/api/routes/vote.go
-24
internal/api/routes/vote.go
···
1
-
package routes
2
-
3
-
import (
4
-
"Coves/internal/api/handlers/vote"
5
-
"Coves/internal/api/middleware"
6
-
"Coves/internal/core/votes"
7
-
8
-
"github.com/go-chi/chi/v5"
9
-
)
10
-
11
-
// RegisterVoteRoutes registers vote-related XRPC endpoints on the router
12
-
// Implements social.coves.interaction.* lexicon endpoints for voting
13
-
func RegisterVoteRoutes(r chi.Router, service votes.Service, authMiddleware *middleware.AtProtoAuthMiddleware) {
14
-
// Initialize handlers
15
-
createVoteHandler := vote.NewCreateVoteHandler(service)
16
-
deleteVoteHandler := vote.NewDeleteVoteHandler(service)
17
-
18
-
// Procedure endpoints (POST) - require authentication
19
-
// social.coves.interaction.createVote - create or toggle a vote on a post/comment
20
-
r.With(authMiddleware.RequireAuth).Post("/xrpc/social.coves.interaction.createVote", createVoteHandler.HandleCreateVote)
21
-
22
-
// social.coves.interaction.deleteVote - delete a vote from a post/comment
23
-
r.With(authMiddleware.RequireAuth).Post("/xrpc/social.coves.interaction.deleteVote", deleteVoteHandler.HandleDeleteVote)
24
-
}
-789
tests/integration/vote_e2e_test.go
-789
tests/integration/vote_e2e_test.go
···
1
-
package integration
2
-
3
-
import (
4
-
"Coves/internal/api/handlers/vote"
5
-
"Coves/internal/api/middleware"
6
-
"Coves/internal/atproto/identity"
7
-
"Coves/internal/atproto/jetstream"
8
-
"Coves/internal/core/communities"
9
-
"Coves/internal/core/posts"
10
-
"Coves/internal/core/users"
11
-
"Coves/internal/core/votes"
12
-
"Coves/internal/db/postgres"
13
-
"bytes"
14
-
"context"
15
-
"database/sql"
16
-
"encoding/json"
17
-
"fmt"
18
-
"net"
19
-
"net/http"
20
-
"net/http/httptest"
21
-
"os"
22
-
"strings"
23
-
"testing"
24
-
"time"
25
-
26
-
"github.com/gorilla/websocket"
27
-
_ "github.com/lib/pq"
28
-
"github.com/pressly/goose/v3"
29
-
"github.com/stretchr/testify/assert"
30
-
"github.com/stretchr/testify/require"
31
-
)
32
-
33
-
// TestVote_E2E_WithJetstream tests the full vote flow with simulated Jetstream:
34
-
// XRPC endpoint → AppView Service → PDS write → (Simulated) Jetstream consumer → DB indexing
35
-
//
36
-
// This is a fast integration test that simulates what happens in production:
37
-
// 1. Client calls POST /xrpc/social.coves.interaction.createVote with auth token
38
-
// 2. Handler validates and calls VoteService.CreateVote()
39
-
// 3. Service writes vote to user's PDS repository
40
-
// 4. (Simulated) PDS broadcasts event to Jetstream
41
-
// 5. Jetstream consumer receives event and indexes vote in AppView DB
42
-
// 6. Vote is now queryable from AppView + post counts updated
43
-
//
44
-
// NOTE: This test simulates the Jetstream event (step 4-5) since we don't have
45
-
// a live PDS/Jetstream in test environment. For true live testing, use TestVote_E2E_LivePDS.
46
-
func TestVote_E2E_WithJetstream(t *testing.T) {
47
-
db := setupTestDB(t)
48
-
defer func() {
49
-
if err := db.Close(); err != nil {
50
-
t.Logf("Failed to close database: %v", err)
51
-
}
52
-
}()
53
-
54
-
// Cleanup old test data first
55
-
_, _ = db.Exec("DELETE FROM votes WHERE voter_did LIKE 'did:plc:votee2e%'")
56
-
_, _ = db.Exec("DELETE FROM posts WHERE community_did = 'did:plc:votecommunity123'")
57
-
_, _ = db.Exec("DELETE FROM communities WHERE did = 'did:plc:votecommunity123'")
58
-
_, _ = db.Exec("DELETE FROM users WHERE did LIKE 'did:plc:votee2e%'")
59
-
60
-
// Setup repositories
61
-
userRepo := postgres.NewUserRepository(db)
62
-
communityRepo := postgres.NewCommunityRepository(db)
63
-
postRepo := postgres.NewPostRepository(db)
64
-
voteRepo := postgres.NewVoteRepository(db)
65
-
66
-
// Setup user service for consumers
67
-
identityConfig := identity.DefaultConfig()
68
-
identityResolver := identity.NewResolver(db, identityConfig)
69
-
userService := users.NewUserService(userRepo, identityResolver, "http://localhost:3001")
70
-
71
-
// Create test users (voter and author)
72
-
voter := createTestUser(t, db, "voter.test", "did:plc:votee2evoter123")
73
-
author := createTestUser(t, db, "author.test", "did:plc:votee2eauthor123")
74
-
75
-
// Create test community
76
-
community := &communities.Community{
77
-
DID: "did:plc:votecommunity123",
78
-
Handle: "votecommunity.test.coves.social",
79
-
Name: "votecommunity",
80
-
DisplayName: "Vote Test Community",
81
-
OwnerDID: "did:plc:votecommunity123",
82
-
CreatedByDID: author.DID,
83
-
HostedByDID: "did:web:coves.test",
84
-
Visibility: "public",
85
-
ModerationType: "moderator",
86
-
RecordURI: "at://did:plc:votecommunity123/social.coves.community.profile/self",
87
-
RecordCID: "fakecid123",
88
-
PDSAccessToken: "fake_token_for_testing",
89
-
PDSRefreshToken: "fake_refresh_token",
90
-
}
91
-
_, err := communityRepo.Create(context.Background(), community)
92
-
if err != nil {
93
-
t.Fatalf("Failed to create test community: %v", err)
94
-
}
95
-
96
-
// Create test post (subject of votes)
97
-
postRkey := generateTID()
98
-
postURI := fmt.Sprintf("at://%s/social.coves.post.record/%s", community.DID, postRkey)
99
-
postCID := "bafy2bzacepostcid123"
100
-
post := &posts.Post{
101
-
URI: postURI,
102
-
CID: postCID,
103
-
RKey: postRkey,
104
-
AuthorDID: author.DID,
105
-
CommunityDID: community.DID,
106
-
Title: stringPtr("Test Post for Voting"),
107
-
Content: stringPtr("This post will receive votes"),
108
-
CreatedAt: time.Now(),
109
-
UpvoteCount: 0,
110
-
DownvoteCount: 0,
111
-
Score: 0,
112
-
}
113
-
err = postRepo.Create(context.Background(), post)
114
-
if err != nil {
115
-
t.Fatalf("Failed to create test post: %v", err)
116
-
}
117
-
118
-
t.Run("Full E2E flow - Create upvote via Jetstream", func(t *testing.T) {
119
-
ctx := context.Background()
120
-
121
-
// STEP 1: Simulate Jetstream consumer receiving a vote CREATE event
122
-
// In real production, this event comes from PDS via Jetstream WebSocket
123
-
voteRkey := generateTID()
124
-
voteURI := fmt.Sprintf("at://%s/social.coves.interaction.vote/%s", voter.DID, voteRkey)
125
-
126
-
jetstreamEvent := jetstream.JetstreamEvent{
127
-
Did: voter.DID, // Vote comes from voter's repo
128
-
Kind: "commit",
129
-
Commit: &jetstream.CommitEvent{
130
-
Operation: "create",
131
-
Collection: "social.coves.interaction.vote",
132
-
RKey: voteRkey,
133
-
CID: "bafy2bzacevotecid123",
134
-
Record: map[string]interface{}{
135
-
"$type": "social.coves.interaction.vote",
136
-
"subject": map[string]interface{}{
137
-
"uri": postURI,
138
-
"cid": postCID,
139
-
},
140
-
"direction": "up",
141
-
"createdAt": time.Now().Format(time.RFC3339),
142
-
},
143
-
},
144
-
}
145
-
146
-
// STEP 2: Process event through Jetstream consumer
147
-
consumer := jetstream.NewVoteEventConsumer(voteRepo, userService, db)
148
-
err := consumer.HandleEvent(ctx, &jetstreamEvent)
149
-
if err != nil {
150
-
t.Fatalf("Jetstream consumer failed to process event: %v", err)
151
-
}
152
-
153
-
// STEP 3: Verify vote was indexed in AppView database
154
-
indexedVote, err := voteRepo.GetByURI(ctx, voteURI)
155
-
if err != nil {
156
-
t.Fatalf("Vote not indexed in AppView: %v", err)
157
-
}
158
-
159
-
// STEP 4: Verify vote fields are correct
160
-
assert.Equal(t, voteURI, indexedVote.URI, "Vote URI should match")
161
-
assert.Equal(t, voter.DID, indexedVote.VoterDID, "Voter DID should match")
162
-
assert.Equal(t, postURI, indexedVote.SubjectURI, "Subject URI should match")
163
-
assert.Equal(t, postCID, indexedVote.SubjectCID, "Subject CID should match (strong reference)")
164
-
assert.Equal(t, "up", indexedVote.Direction, "Direction should be 'up'")
165
-
166
-
// STEP 5: Verify post vote counts were updated atomically
167
-
updatedPost, err := postRepo.GetByURI(ctx, postURI)
168
-
require.NoError(t, err, "Post should still exist")
169
-
assert.Equal(t, 1, updatedPost.UpvoteCount, "Post upvote_count should be 1")
170
-
assert.Equal(t, 0, updatedPost.DownvoteCount, "Post downvote_count should be 0")
171
-
assert.Equal(t, 1, updatedPost.Score, "Post score should be 1 (upvotes - downvotes)")
172
-
173
-
t.Logf("✓ E2E test passed! Vote indexed with URI: %s, post upvotes: %d", indexedVote.URI, updatedPost.UpvoteCount)
174
-
})
175
-
176
-
t.Run("Create downvote and verify counts", func(t *testing.T) {
177
-
ctx := context.Background()
178
-
179
-
// Create a different voter for this test to avoid unique constraint violation
180
-
downvoter := createTestUser(t, db, "downvoter.test", "did:plc:votee2edownvoter")
181
-
182
-
// Create downvote
183
-
voteRkey := generateTID()
184
-
voteURI := fmt.Sprintf("at://%s/social.coves.interaction.vote/%s", downvoter.DID, voteRkey)
185
-
186
-
jetstreamEvent := jetstream.JetstreamEvent{
187
-
Did: downvoter.DID,
188
-
Kind: "commit",
189
-
Commit: &jetstream.CommitEvent{
190
-
Operation: "create",
191
-
Collection: "social.coves.interaction.vote",
192
-
RKey: voteRkey,
193
-
CID: "bafy2bzacedownvotecid",
194
-
Record: map[string]interface{}{
195
-
"$type": "social.coves.interaction.vote",
196
-
"subject": map[string]interface{}{
197
-
"uri": postURI,
198
-
"cid": postCID,
199
-
},
200
-
"direction": "down",
201
-
"createdAt": time.Now().Format(time.RFC3339),
202
-
},
203
-
},
204
-
}
205
-
206
-
consumer := jetstream.NewVoteEventConsumer(voteRepo, userService, db)
207
-
err := consumer.HandleEvent(ctx, &jetstreamEvent)
208
-
require.NoError(t, err, "Consumer should process downvote")
209
-
210
-
// Verify vote indexed
211
-
indexedVote, err := voteRepo.GetByURI(ctx, voteURI)
212
-
require.NoError(t, err, "Downvote should be indexed")
213
-
assert.Equal(t, "down", indexedVote.Direction, "Direction should be 'down'")
214
-
215
-
// Verify post counts (now has 1 upvote + 1 downvote from previous test)
216
-
updatedPost, err := postRepo.GetByURI(ctx, postURI)
217
-
require.NoError(t, err)
218
-
assert.Equal(t, 1, updatedPost.UpvoteCount, "Upvote count should still be 1")
219
-
assert.Equal(t, 1, updatedPost.DownvoteCount, "Downvote count should be 1")
220
-
assert.Equal(t, 0, updatedPost.Score, "Score should be 0 (1 up - 1 down)")
221
-
222
-
t.Logf("✓ Downvote indexed, post counts: up=%d down=%d score=%d",
223
-
updatedPost.UpvoteCount, updatedPost.DownvoteCount, updatedPost.Score)
224
-
})
225
-
226
-
t.Run("Delete vote and verify counts decremented", func(t *testing.T) {
227
-
ctx := context.Background()
228
-
229
-
// Create a different voter for this test
230
-
deletevoter := createTestUser(t, db, "deletevoter.test", "did:plc:votee2edeletevoter")
231
-
232
-
// Get current counts
233
-
beforePost, _ := postRepo.GetByURI(ctx, postURI)
234
-
235
-
// Create a vote first
236
-
voteRkey := generateTID()
237
-
voteURI := fmt.Sprintf("at://%s/social.coves.interaction.vote/%s", deletevoter.DID, voteRkey)
238
-
239
-
createEvent := jetstream.JetstreamEvent{
240
-
Did: deletevoter.DID,
241
-
Kind: "commit",
242
-
Commit: &jetstream.CommitEvent{
243
-
Operation: "create",
244
-
Collection: "social.coves.interaction.vote",
245
-
RKey: voteRkey,
246
-
CID: "bafy2bzacedeleteme",
247
-
Record: map[string]interface{}{
248
-
"$type": "social.coves.interaction.vote",
249
-
"subject": map[string]interface{}{
250
-
"uri": postURI,
251
-
"cid": postCID,
252
-
},
253
-
"direction": "up",
254
-
"createdAt": time.Now().Format(time.RFC3339),
255
-
},
256
-
},
257
-
}
258
-
259
-
consumer := jetstream.NewVoteEventConsumer(voteRepo, userService, db)
260
-
err := consumer.HandleEvent(ctx, &createEvent)
261
-
require.NoError(t, err)
262
-
263
-
// Now delete it
264
-
deleteEvent := jetstream.JetstreamEvent{
265
-
Did: deletevoter.DID,
266
-
Kind: "commit",
267
-
Commit: &jetstream.CommitEvent{
268
-
Operation: "delete",
269
-
Collection: "social.coves.interaction.vote",
270
-
RKey: voteRkey,
271
-
},
272
-
}
273
-
274
-
err = consumer.HandleEvent(ctx, &deleteEvent)
275
-
require.NoError(t, err, "Consumer should process delete")
276
-
277
-
// Verify vote is soft-deleted
278
-
deletedVote, err := voteRepo.GetByURI(ctx, voteURI)
279
-
require.NoError(t, err, "Vote should still exist (soft delete)")
280
-
assert.NotNil(t, deletedVote.DeletedAt, "Vote should have deleted_at timestamp")
281
-
282
-
// Verify post counts decremented
283
-
afterPost, err := postRepo.GetByURI(ctx, postURI)
284
-
require.NoError(t, err)
285
-
assert.Equal(t, beforePost.UpvoteCount, afterPost.UpvoteCount,
286
-
"Upvote count should be back to original (delete decremented)")
287
-
288
-
t.Logf("✓ Vote deleted, counts decremented correctly")
289
-
})
290
-
291
-
t.Run("Idempotent indexing - duplicate events", func(t *testing.T) {
292
-
ctx := context.Background()
293
-
294
-
// Create a different voter for this test
295
-
idempotentvoter := createTestUser(t, db, "idempotentvoter.test", "did:plc:votee2eidempotent")
296
-
297
-
// Create a vote
298
-
voteRkey := generateTID()
299
-
voteURI := fmt.Sprintf("at://%s/social.coves.interaction.vote/%s", idempotentvoter.DID, voteRkey)
300
-
301
-
event := jetstream.JetstreamEvent{
302
-
Did: idempotentvoter.DID,
303
-
Kind: "commit",
304
-
Commit: &jetstream.CommitEvent{
305
-
Operation: "create",
306
-
Collection: "social.coves.interaction.vote",
307
-
RKey: voteRkey,
308
-
CID: "bafy2bzaceidempotent",
309
-
Record: map[string]interface{}{
310
-
"$type": "social.coves.interaction.vote",
311
-
"subject": map[string]interface{}{
312
-
"uri": postURI,
313
-
"cid": postCID,
314
-
},
315
-
"direction": "up",
316
-
"createdAt": time.Now().Format(time.RFC3339),
317
-
},
318
-
},
319
-
}
320
-
321
-
consumer := jetstream.NewVoteEventConsumer(voteRepo, userService, db)
322
-
323
-
// First event - should succeed
324
-
err := consumer.HandleEvent(ctx, &event)
325
-
require.NoError(t, err, "First event should succeed")
326
-
327
-
// Get counts after first event
328
-
firstPost, _ := postRepo.GetByURI(ctx, postURI)
329
-
330
-
// Second event (duplicate) - should be handled gracefully
331
-
err = consumer.HandleEvent(ctx, &event)
332
-
require.NoError(t, err, "Duplicate event should be handled gracefully")
333
-
334
-
// Verify counts NOT incremented again (idempotent)
335
-
secondPost, err := postRepo.GetByURI(ctx, postURI)
336
-
require.NoError(t, err)
337
-
assert.Equal(t, firstPost.UpvoteCount, secondPost.UpvoteCount,
338
-
"Duplicate event should not increment count again")
339
-
340
-
// Verify only one vote in database
341
-
vote, err := voteRepo.GetByURI(ctx, voteURI)
342
-
require.NoError(t, err)
343
-
assert.Equal(t, voteURI, vote.URI, "Should still be the same vote")
344
-
345
-
t.Logf("✓ Idempotency test passed - duplicate event handled correctly")
346
-
})
347
-
348
-
t.Run("Security: Vote from wrong repository rejected", func(t *testing.T) {
349
-
ctx := context.Background()
350
-
351
-
// SECURITY TEST: Try to create a vote that claims to be from the voter
352
-
// but actually comes from a different user's repository
353
-
// This should be REJECTED by the consumer
354
-
355
-
maliciousUser := createTestUser(t, db, "hacker.test", "did:plc:hacker123")
356
-
357
-
maliciousEvent := jetstream.JetstreamEvent{
358
-
Did: maliciousUser.DID, // Event from hacker's repo
359
-
Kind: "commit",
360
-
Commit: &jetstream.CommitEvent{
361
-
Operation: "create",
362
-
Collection: "social.coves.interaction.vote",
363
-
RKey: generateTID(),
364
-
CID: "bafy2bzacefake",
365
-
Record: map[string]interface{}{
366
-
"$type": "social.coves.interaction.vote",
367
-
"subject": map[string]interface{}{
368
-
"uri": postURI,
369
-
"cid": postCID,
370
-
},
371
-
"direction": "up",
372
-
"createdAt": time.Now().Format(time.RFC3339),
373
-
},
374
-
},
375
-
}
376
-
377
-
consumer := jetstream.NewVoteEventConsumer(voteRepo, userService, db)
378
-
err := consumer.HandleEvent(ctx, &maliciousEvent)
379
-
380
-
// Should succeed (vote is created in hacker's repo, which is valid)
381
-
// The vote record itself is FROM their repo, so it's legitimate
382
-
// This is different from posts which must come from community repo
383
-
assert.NoError(t, err, "Votes in user repos are valid")
384
-
385
-
t.Logf("✓ Security validation passed - user repo votes are allowed")
386
-
})
387
-
}
388
-
389
-
// TestVote_E2E_LivePDS tests the COMPLETE end-to-end flow with a live PDS:
390
-
// 1. HTTP POST to /xrpc/social.coves.interaction.createVote (with auth)
391
-
// 2. Handler → Service → Write to user's PDS repository
392
-
// 3. PDS → Jetstream firehose event
393
-
// 4. Jetstream consumer → Index in AppView database
394
-
// 5. Verify vote appears in database + post counts updated
395
-
//
396
-
// This is a TRUE E2E test that requires:
397
-
// - Live PDS running at PDS_URL (default: http://localhost:3001)
398
-
// - Live Jetstream running at JETSTREAM_URL (default: ws://localhost:6008/subscribe)
399
-
// - Test database running
400
-
func TestVote_E2E_LivePDS(t *testing.T) {
401
-
if testing.Short() {
402
-
t.Skip("Skipping live PDS E2E test in short mode")
403
-
}
404
-
405
-
// Setup test database
406
-
dbURL := os.Getenv("TEST_DATABASE_URL")
407
-
if dbURL == "" {
408
-
dbURL = "postgres://test_user:test_password@localhost:5434/coves_test?sslmode=disable"
409
-
}
410
-
411
-
db, err := sql.Open("postgres", dbURL)
412
-
require.NoError(t, err, "Failed to connect to test database")
413
-
defer func() {
414
-
if closeErr := db.Close(); closeErr != nil {
415
-
t.Logf("Failed to close database: %v", closeErr)
416
-
}
417
-
}()
418
-
419
-
// Run migrations
420
-
require.NoError(t, goose.SetDialect("postgres"))
421
-
require.NoError(t, goose.Up(db, "../../internal/db/migrations"))
422
-
423
-
// Check if PDS is running
424
-
pdsURL := os.Getenv("PDS_URL")
425
-
if pdsURL == "" {
426
-
pdsURL = "http://localhost:3001"
427
-
}
428
-
429
-
healthResp, err := http.Get(pdsURL + "/xrpc/_health")
430
-
if err != nil {
431
-
t.Skipf("PDS not running at %s: %v", pdsURL, err)
432
-
}
433
-
_ = healthResp.Body.Close()
434
-
435
-
// Check if Jetstream is running
436
-
jetstreamHealthURL := "http://127.0.0.1:6009/metrics" // Use 127.0.0.1 for IPv4
437
-
jetstreamResp, err := http.Get(jetstreamHealthURL)
438
-
if err != nil {
439
-
t.Skipf("Jetstream not running: %v", err)
440
-
}
441
-
_ = jetstreamResp.Body.Close()
442
-
443
-
ctx := context.Background()
444
-
445
-
// Cleanup old test data
446
-
_, _ = db.Exec("DELETE FROM votes WHERE voter_did LIKE 'did:plc:votee2elive%' OR voter_did IN (SELECT did FROM users WHERE handle LIKE '%votee2elive%')")
447
-
_, _ = db.Exec("DELETE FROM posts WHERE community_did LIKE 'did:plc:votee2elive%'")
448
-
_, _ = db.Exec("DELETE FROM communities WHERE did LIKE 'did:plc:votee2elive%'")
449
-
_, _ = db.Exec("DELETE FROM users WHERE did LIKE 'did:plc:votee2elive%' OR handle LIKE '%votee2elive%' OR handle LIKE '%authore2e%'")
450
-
451
-
// Setup repositories and services
452
-
userRepo := postgres.NewUserRepository(db)
453
-
communityRepo := postgres.NewCommunityRepository(db)
454
-
postRepo := postgres.NewPostRepository(db)
455
-
voteRepo := postgres.NewVoteRepository(db)
456
-
457
-
identityConfig := identity.DefaultConfig()
458
-
identityResolver := identity.NewResolver(db, identityConfig)
459
-
userService := users.NewUserService(userRepo, identityResolver, pdsURL)
460
-
461
-
// Create test voter
462
-
voter := createTestUser(t, db, "votee2elive.bsky.social", "did:plc:votee2elive123")
463
-
464
-
// Create test community and post (simplified - using fake credentials)
465
-
author := createTestUser(t, db, "authore2e.bsky.social", "did:plc:votee2eliveauthor")
466
-
community := &communities.Community{
467
-
DID: "did:plc:votee2elivecommunity",
468
-
Handle: "votee2elivecommunity.test.coves.social",
469
-
Name: "votee2elivecommunity",
470
-
DisplayName: "Vote E2E Live Community",
471
-
OwnerDID: author.DID,
472
-
CreatedByDID: author.DID,
473
-
HostedByDID: "did:web:coves.test",
474
-
Visibility: "public",
475
-
ModerationType: "moderator",
476
-
RecordURI: "at://did:plc:votee2elivecommunity/social.coves.community.profile/self",
477
-
RecordCID: "fakecid",
478
-
PDSAccessToken: "fake_token",
479
-
PDSRefreshToken: "fake_refresh",
480
-
}
481
-
_, err = communityRepo.Create(ctx, community)
482
-
require.NoError(t, err)
483
-
484
-
postRkey := generateTID()
485
-
postURI := fmt.Sprintf("at://%s/social.coves.post.record/%s", community.DID, postRkey)
486
-
postCID := "bafy2bzaceposte2e"
487
-
post := &posts.Post{
488
-
URI: postURI,
489
-
CID: postCID,
490
-
RKey: postRkey,
491
-
AuthorDID: author.DID,
492
-
CommunityDID: community.DID,
493
-
Title: stringPtr("E2E Vote Test Post"),
494
-
Content: stringPtr("This post will receive live votes"),
495
-
CreatedAt: time.Now(),
496
-
UpvoteCount: 0,
497
-
DownvoteCount: 0,
498
-
Score: 0,
499
-
}
500
-
err = postRepo.Create(ctx, post)
501
-
require.NoError(t, err)
502
-
503
-
// Setup vote service and handler
504
-
voteService := votes.NewVoteService(voteRepo, postRepo, pdsURL)
505
-
voteHandler := vote.NewCreateVoteHandler(voteService)
506
-
authMiddleware := middleware.NewAtProtoAuthMiddleware(nil, true) // Skip JWT verification for testing
507
-
508
-
t.Run("Live E2E: Create vote and verify via Jetstream", func(t *testing.T) {
509
-
t.Logf("\n🔄 TRUE E2E: Creating vote via XRPC endpoint...")
510
-
511
-
// Authenticate voter with PDS to get real access token
512
-
// Note: This assumes the voter account already exists on PDS
513
-
// For a complete test, you'd create the account first via com.atproto.server.createAccount
514
-
instanceHandle := os.Getenv("PDS_INSTANCE_HANDLE")
515
-
instancePassword := os.Getenv("PDS_INSTANCE_PASSWORD")
516
-
if instanceHandle == "" {
517
-
instanceHandle = "testuser123.local.coves.dev"
518
-
}
519
-
if instancePassword == "" {
520
-
instancePassword = "test-password-123"
521
-
}
522
-
523
-
t.Logf("🔐 Authenticating voter with PDS as: %s", instanceHandle)
524
-
voterAccessToken, voterDID, err := authenticateWithPDS(pdsURL, instanceHandle, instancePassword)
525
-
if err != nil {
526
-
t.Skipf("Failed to authenticate voter with PDS (account may not exist): %v", err)
527
-
}
528
-
t.Logf("✅ Authenticated - Voter DID: %s", voterDID)
529
-
530
-
// Update voter record to match authenticated DID
531
-
_, err = db.Exec("UPDATE users SET did = $1 WHERE did = $2", voterDID, voter.DID)
532
-
require.NoError(t, err)
533
-
voter.DID = voterDID
534
-
535
-
// Build HTTP request for vote creation
536
-
reqBody := map[string]interface{}{
537
-
"subject": postURI,
538
-
"direction": "up",
539
-
}
540
-
reqJSON, err := json.Marshal(reqBody)
541
-
require.NoError(t, err)
542
-
543
-
// Create HTTP request
544
-
req := httptest.NewRequest("POST", "/xrpc/social.coves.interaction.createVote", bytes.NewReader(reqJSON))
545
-
req.Header.Set("Content-Type", "application/json")
546
-
547
-
// Use REAL PDS access token (not mock JWT)
548
-
req.Header.Set("Authorization", "Bearer "+voterAccessToken)
549
-
550
-
// Execute request through auth middleware + handler
551
-
rr := httptest.NewRecorder()
552
-
handler := authMiddleware.RequireAuth(http.HandlerFunc(voteHandler.HandleCreateVote))
553
-
handler.ServeHTTP(rr, req)
554
-
555
-
// Check response
556
-
require.Equal(t, http.StatusOK, rr.Code, "Handler should return 200 OK, body: %s", rr.Body.String())
557
-
558
-
// Parse response
559
-
var response map[string]interface{}
560
-
err = json.NewDecoder(rr.Body).Decode(&response)
561
-
require.NoError(t, err, "Failed to parse response")
562
-
563
-
voteURI := response["uri"].(string)
564
-
voteCID := response["cid"].(string)
565
-
566
-
t.Logf("✅ Vote created on PDS:")
567
-
t.Logf(" URI: %s", voteURI)
568
-
t.Logf(" CID: %s", voteCID)
569
-
570
-
// ====================================================================================
571
-
// Part 2: Query the PDS to verify the vote record exists
572
-
// ====================================================================================
573
-
t.Run("2a. Verify vote record on PDS", func(t *testing.T) {
574
-
t.Logf("\n📡 Querying PDS for vote record...")
575
-
576
-
// Extract rkey from vote URI (at://did/collection/rkey)
577
-
parts := strings.Split(voteURI, "/")
578
-
rkey := parts[len(parts)-1]
579
-
580
-
// Query PDS for the vote record
581
-
getRecordURL := fmt.Sprintf("%s/xrpc/com.atproto.repo.getRecord?repo=%s&collection=%s&rkey=%s",
582
-
pdsURL, voterDID, "social.coves.interaction.vote", rkey)
583
-
584
-
t.Logf(" GET %s", getRecordURL)
585
-
586
-
pdsResp, err := http.Get(getRecordURL)
587
-
require.NoError(t, err, "Failed to query PDS")
588
-
defer pdsResp.Body.Close()
589
-
590
-
require.Equal(t, http.StatusOK, pdsResp.StatusCode, "Vote record should exist on PDS")
591
-
592
-
var pdsRecord struct {
593
-
Value map[string]interface{} `json:"value"`
594
-
URI string `json:"uri"`
595
-
CID string `json:"cid"`
596
-
}
597
-
598
-
err = json.NewDecoder(pdsResp.Body).Decode(&pdsRecord)
599
-
require.NoError(t, err, "Failed to decode PDS response")
600
-
601
-
t.Logf("✅ Vote record found on PDS!")
602
-
t.Logf(" URI: %s", pdsRecord.URI)
603
-
t.Logf(" CID: %s", pdsRecord.CID)
604
-
t.Logf(" Direction: %v", pdsRecord.Value["direction"])
605
-
t.Logf(" Subject: %v", pdsRecord.Value["subject"])
606
-
607
-
// Verify the record matches what we created
608
-
assert.Equal(t, voteURI, pdsRecord.URI, "PDS URI should match")
609
-
assert.Equal(t, voteCID, pdsRecord.CID, "PDS CID should match")
610
-
assert.Equal(t, "up", pdsRecord.Value["direction"], "Direction should be 'up'")
611
-
612
-
// Print full record for inspection
613
-
recordJSON, _ := json.MarshalIndent(pdsRecord.Value, " ", " ")
614
-
t.Logf(" Full record:\n %s", string(recordJSON))
615
-
})
616
-
617
-
// ====================================================================================
618
-
// Part 2b: TRUE E2E - Real Jetstream Firehose Consumer
619
-
// ====================================================================================
620
-
t.Run("2b. Real Jetstream Firehose Consumption", func(t *testing.T) {
621
-
t.Logf("\n🔄 TRUE E2E: Subscribing to real Jetstream firehose...")
622
-
623
-
// Get PDS hostname for Jetstream filtering
624
-
pdsHostname := strings.TrimPrefix(pdsURL, "http://")
625
-
pdsHostname = strings.TrimPrefix(pdsHostname, "https://")
626
-
pdsHostname = strings.Split(pdsHostname, ":")[0] // Remove port
627
-
628
-
// Build Jetstream URL with filters for vote records
629
-
jetstreamURL := fmt.Sprintf("ws://%s:6008/subscribe?wantedCollections=social.coves.interaction.vote",
630
-
pdsHostname)
631
-
632
-
t.Logf(" Jetstream URL: %s", jetstreamURL)
633
-
t.Logf(" Looking for vote URI: %s", voteURI)
634
-
t.Logf(" Voter DID: %s", voterDID)
635
-
636
-
// Create vote consumer (same as main.go)
637
-
consumer := jetstream.NewVoteEventConsumer(voteRepo, userService, db)
638
-
639
-
// Channels to receive the event
640
-
eventChan := make(chan *jetstream.JetstreamEvent, 10)
641
-
errorChan := make(chan error, 1)
642
-
done := make(chan bool)
643
-
644
-
// Start Jetstream WebSocket subscriber in background
645
-
go func() {
646
-
err := subscribeToJetstreamForVote(ctx, jetstreamURL, voterDID, postURI, consumer, eventChan, errorChan, done)
647
-
if err != nil {
648
-
errorChan <- err
649
-
}
650
-
}()
651
-
652
-
// Wait for event or timeout
653
-
t.Logf("⏳ Waiting for Jetstream event (max 30 seconds)...")
654
-
655
-
select {
656
-
case event := <-eventChan:
657
-
t.Logf("✅ Received real Jetstream event!")
658
-
t.Logf(" Event DID: %s", event.Did)
659
-
t.Logf(" Collection: %s", event.Commit.Collection)
660
-
t.Logf(" Operation: %s", event.Commit.Operation)
661
-
t.Logf(" RKey: %s", event.Commit.RKey)
662
-
663
-
// Verify it's for our voter
664
-
assert.Equal(t, voterDID, event.Did, "Event should be from voter's repo")
665
-
666
-
// Verify vote was indexed in AppView database
667
-
t.Logf("\n🔍 Querying AppView database for indexed vote...")
668
-
669
-
indexedVote, err := voteRepo.GetByVoterAndSubject(ctx, voterDID, postURI)
670
-
require.NoError(t, err, "Vote should be indexed in AppView")
671
-
672
-
t.Logf("✅ Vote indexed in AppView:")
673
-
t.Logf(" URI: %s", indexedVote.URI)
674
-
t.Logf(" CID: %s", indexedVote.CID)
675
-
t.Logf(" Voter DID: %s", indexedVote.VoterDID)
676
-
t.Logf(" Subject: %s", indexedVote.SubjectURI)
677
-
t.Logf(" Direction: %s", indexedVote.Direction)
678
-
679
-
// Verify all fields match
680
-
assert.Equal(t, voteURI, indexedVote.URI, "URI should match")
681
-
assert.Equal(t, voteCID, indexedVote.CID, "CID should match")
682
-
assert.Equal(t, voterDID, indexedVote.VoterDID, "Voter DID should match")
683
-
assert.Equal(t, postURI, indexedVote.SubjectURI, "Subject URI should match")
684
-
assert.Equal(t, "up", indexedVote.Direction, "Direction should be 'up'")
685
-
686
-
// Verify post counts were updated
687
-
t.Logf("\n🔍 Verifying post vote counts updated...")
688
-
updatedPost, err := postRepo.GetByURI(ctx, postURI)
689
-
require.NoError(t, err, "Post should exist")
690
-
691
-
t.Logf("✅ Post vote counts updated:")
692
-
t.Logf(" Upvotes: %d", updatedPost.UpvoteCount)
693
-
t.Logf(" Downvotes: %d", updatedPost.DownvoteCount)
694
-
t.Logf(" Score: %d", updatedPost.Score)
695
-
696
-
assert.Equal(t, 1, updatedPost.UpvoteCount, "Upvote count should be 1")
697
-
assert.Equal(t, 0, updatedPost.DownvoteCount, "Downvote count should be 0")
698
-
assert.Equal(t, 1, updatedPost.Score, "Score should be 1")
699
-
700
-
// Signal to stop Jetstream consumer
701
-
close(done)
702
-
703
-
t.Log("\n✅ TRUE E2E COMPLETE: PDS → Jetstream → Consumer → AppView ✓")
704
-
705
-
case err := <-errorChan:
706
-
t.Fatalf("❌ Jetstream error: %v", err)
707
-
708
-
case <-time.After(30 * time.Second):
709
-
t.Fatalf("❌ Timeout: No Jetstream event received within 30 seconds")
710
-
}
711
-
})
712
-
})
713
-
}
714
-
715
-
// subscribeToJetstreamForVote subscribes to real Jetstream firehose and processes vote events
716
-
// This helper creates a WebSocket connection to Jetstream and waits for vote events
717
-
func subscribeToJetstreamForVote(
718
-
ctx context.Context,
719
-
jetstreamURL string,
720
-
targetVoterDID string,
721
-
targetSubjectURI string,
722
-
consumer *jetstream.VoteEventConsumer,
723
-
eventChan chan<- *jetstream.JetstreamEvent,
724
-
errorChan chan<- error,
725
-
done <-chan bool,
726
-
) error {
727
-
conn, _, err := websocket.DefaultDialer.Dial(jetstreamURL, nil)
728
-
if err != nil {
729
-
return fmt.Errorf("failed to connect to Jetstream: %w", err)
730
-
}
731
-
defer func() { _ = conn.Close() }()
732
-
733
-
// Read messages until we find our event or receive done signal
734
-
for {
735
-
select {
736
-
case <-done:
737
-
return nil
738
-
case <-ctx.Done():
739
-
return ctx.Err()
740
-
default:
741
-
// Set read deadline to avoid blocking forever
742
-
if err := conn.SetReadDeadline(time.Now().Add(5 * time.Second)); err != nil {
743
-
return fmt.Errorf("failed to set read deadline: %w", err)
744
-
}
745
-
746
-
var event jetstream.JetstreamEvent
747
-
err := conn.ReadJSON(&event)
748
-
if err != nil {
749
-
// Check if it's a timeout (expected)
750
-
if websocket.IsCloseError(err, websocket.CloseNormalClosure) {
751
-
return nil
752
-
}
753
-
if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
754
-
continue // Timeout is expected, keep listening
755
-
}
756
-
return fmt.Errorf("failed to read Jetstream message: %w", err)
757
-
}
758
-
759
-
// Check if this is a vote event for the target voter + subject
760
-
if event.Did == targetVoterDID && event.Kind == "commit" &&
761
-
event.Commit != nil && event.Commit.Collection == "social.coves.interaction.vote" {
762
-
763
-
// Verify it's for the target subject
764
-
record := event.Commit.Record
765
-
if subject, ok := record["subject"].(map[string]interface{}); ok {
766
-
if subjectURI, ok := subject["uri"].(string); ok && subjectURI == targetSubjectURI {
767
-
// This is our vote! Process it
768
-
if err := consumer.HandleEvent(ctx, &event); err != nil {
769
-
return fmt.Errorf("failed to process event: %w", err)
770
-
}
771
-
772
-
// Send to channel so test can verify
773
-
select {
774
-
case eventChan <- &event:
775
-
return nil
776
-
case <-time.After(1 * time.Second):
777
-
return fmt.Errorf("timeout sending event to channel")
778
-
}
779
-
}
780
-
}
781
-
}
782
-
}
783
-
}
784
-
}
785
-
786
-
// Helper function
787
-
func stringPtr(s string) *string {
788
-
return &s
789
-
}