···
14
+
"github.com/bluesky-social/indigo/atproto/auth/oauth"
15
+
"github.com/bluesky-social/indigo/atproto/syntax"
17
+
oauthclient "Coves/internal/atproto/oauth"
20
+
// voteService implements the Service interface for vote operations
21
+
type voteService struct {
23
+
oauthClient *oauthclient.OAuthClient
24
+
oauthStore oauth.ClientAuthStore
28
+
// NewService creates a new vote service instance
29
+
func NewService(repo Repository, oauthClient *oauthclient.OAuthClient, oauthStore oauth.ClientAuthStore, logger *slog.Logger) Service {
31
+
logger = slog.Default()
33
+
return &voteService{
35
+
oauthClient: oauthClient,
36
+
oauthStore: oauthStore,
41
+
// CreateVote creates a new vote or toggles off an existing vote
42
+
// Implements the toggle behavior:
43
+
// - No existing vote → Create new vote with given direction
44
+
// - Vote exists with same direction → Delete vote (toggle off)
45
+
// - Vote exists with different direction → Update to new direction
46
+
func (s *voteService) CreateVote(ctx context.Context, session *oauth.ClientSessionData, req CreateVoteRequest) (*CreateVoteResponse, error) {
47
+
// Validate direction
48
+
if req.Direction != "up" && req.Direction != "down" {
49
+
return nil, ErrInvalidDirection
52
+
// Validate subject URI format
53
+
if req.Subject.URI == "" {
54
+
return nil, ErrInvalidSubject
56
+
if !strings.HasPrefix(req.Subject.URI, "at://") {
57
+
return nil, ErrInvalidSubject
60
+
// Validate subject CID is provided
61
+
if req.Subject.CID == "" {
62
+
return nil, ErrInvalidSubject
65
+
// Check for existing vote by querying PDS directly (source of truth)
66
+
// This avoids eventual consistency issues with the AppView database
67
+
existing, err := s.getVoteFromPDS(ctx, session, req.Subject.URI)
69
+
s.logger.Error("failed to check existing vote on PDS",
71
+
"voter", session.AccountDID,
72
+
"subject", req.Subject.URI)
73
+
return nil, fmt.Errorf("failed to check existing vote: %w", err)
77
+
if existing != nil {
78
+
// Vote exists - check if same direction
79
+
if existing.Direction == req.Direction {
80
+
// Same direction - toggle off (delete)
81
+
if err := s.deleteVoteRecord(ctx, session, existing.RKey); err != nil {
82
+
s.logger.Error("failed to delete vote on PDS",
84
+
"voter", session.AccountDID,
85
+
"rkey", existing.RKey)
86
+
return nil, fmt.Errorf("failed to delete vote: %w", err)
89
+
s.logger.Info("vote toggled off",
90
+
"voter", session.AccountDID,
91
+
"subject", req.Subject.URI,
92
+
"direction", req.Direction)
94
+
// Return empty response to indicate deletion
95
+
return &CreateVoteResponse{
101
+
// Different direction - delete old vote first, then create new one
102
+
if err := s.deleteVoteRecord(ctx, session, existing.RKey); err != nil {
103
+
s.logger.Error("failed to delete existing vote on PDS",
105
+
"voter", session.AccountDID,
106
+
"rkey", existing.RKey)
107
+
return nil, fmt.Errorf("failed to delete existing vote: %w", err)
110
+
s.logger.Info("deleted existing vote before creating new direction",
111
+
"voter", session.AccountDID,
112
+
"subject", req.Subject.URI,
113
+
"old_direction", existing.Direction,
114
+
"new_direction", req.Direction)
118
+
uri, cid, err := s.createVoteRecord(ctx, session, req)
120
+
s.logger.Error("failed to create vote on PDS",
122
+
"voter", session.AccountDID,
123
+
"subject", req.Subject.URI,
124
+
"direction", req.Direction)
125
+
return nil, fmt.Errorf("failed to create vote: %w", err)
128
+
s.logger.Info("vote created",
129
+
"voter", session.AccountDID,
130
+
"subject", req.Subject.URI,
131
+
"direction", req.Direction,
135
+
return &CreateVoteResponse{
141
+
// DeleteVote removes a vote on the specified subject
142
+
func (s *voteService) DeleteVote(ctx context.Context, session *oauth.ClientSessionData, req DeleteVoteRequest) error {
143
+
// Validate subject URI format
144
+
if req.Subject.URI == "" {
145
+
return ErrInvalidSubject
147
+
if !strings.HasPrefix(req.Subject.URI, "at://") {
148
+
return ErrInvalidSubject
151
+
// Find existing vote by querying PDS directly (source of truth)
152
+
// This avoids eventual consistency issues with the AppView database
153
+
existing, err := s.getVoteFromPDS(ctx, session, req.Subject.URI)
155
+
s.logger.Error("failed to find vote on PDS",
157
+
"voter", session.AccountDID,
158
+
"subject", req.Subject.URI)
159
+
return fmt.Errorf("failed to find vote: %w", err)
161
+
if existing == nil {
162
+
return ErrVoteNotFound
165
+
// Delete the vote record from user's PDS
166
+
if err := s.deleteVoteRecord(ctx, session, existing.RKey); err != nil {
167
+
s.logger.Error("failed to delete vote on PDS",
169
+
"voter", session.AccountDID,
170
+
"rkey", existing.RKey)
171
+
return fmt.Errorf("failed to delete vote: %w", err)
174
+
s.logger.Info("vote deleted",
175
+
"voter", session.AccountDID,
176
+
"subject", req.Subject.URI,
177
+
"uri", existing.URI)
182
+
// createVoteRecord writes a vote record to the user's PDS
183
+
func (s *voteService) createVoteRecord(ctx context.Context, session *oauth.ClientSessionData, req CreateVoteRequest) (string, string, error) {
184
+
// Generate TID for the record key
185
+
tid := syntax.NewTIDNow(0)
187
+
// Build vote record following the lexicon schema
188
+
record := VoteRecord{
189
+
Type: "social.coves.feed.vote",
190
+
Subject: StrongRef{
191
+
URI: req.Subject.URI,
192
+
CID: req.Subject.CID,
194
+
Direction: req.Direction,
195
+
CreatedAt: time.Now().UTC().Format(time.RFC3339),
198
+
// Call com.atproto.repo.createRecord on the user's PDS
199
+
endpoint := fmt.Sprintf("%s/xrpc/com.atproto.repo.createRecord", strings.TrimSuffix(session.HostURL, "/"))
201
+
payload := map[string]interface{}{
202
+
"repo": session.AccountDID.String(),
203
+
"collection": "social.coves.feed.vote",
204
+
"rkey": tid.String(),
208
+
uri, cid, err := s.callPDSWithAuth(ctx, "POST", endpoint, payload, session.AccessToken)
213
+
return uri, cid, nil
216
+
// getVoteFromPDS queries the user's PDS directly to find an existing vote for a subject.
217
+
// This avoids eventual consistency issues with the AppView database populated by Jetstream.
218
+
// Paginates through all vote records to handle users with >100 votes.
219
+
// Returns the vote record with rkey, or nil if no vote exists for the subject.
220
+
func (s *voteService) getVoteFromPDS(ctx context.Context, session *oauth.ClientSessionData, subjectURI string) (*existingVote, error) {
221
+
baseURL := fmt.Sprintf("%s/xrpc/com.atproto.repo.listRecords?repo=%s&collection=social.coves.feed.vote&limit=100",
222
+
strings.TrimSuffix(session.HostURL, "/"),
223
+
session.AccountDID.String())
225
+
client := &http.Client{Timeout: 10 * time.Second}
228
+
// Paginate through all vote records
230
+
endpoint := baseURL
232
+
endpoint = fmt.Sprintf("%s&cursor=%s", baseURL, cursor)
235
+
req, err := http.NewRequestWithContext(ctx, "GET", endpoint, nil)
237
+
return nil, fmt.Errorf("failed to create request: %w", err)
239
+
req.Header.Set("Authorization", "Bearer "+session.AccessToken)
241
+
resp, err := client.Do(req)
243
+
return nil, fmt.Errorf("failed to call PDS: %w", err)
246
+
body, err := io.ReadAll(resp.Body)
247
+
closeErr := resp.Body.Close()
248
+
if closeErr != nil {
249
+
s.logger.Warn("failed to close response body", "error", closeErr)
252
+
return nil, fmt.Errorf("failed to read response: %w", err)
255
+
// Handle auth errors - map to ErrNotAuthorized per lexicon
256
+
if resp.StatusCode == http.StatusUnauthorized || resp.StatusCode == http.StatusForbidden {
257
+
s.logger.Warn("PDS auth failure",
258
+
"status", resp.StatusCode,
259
+
"did", session.AccountDID)
260
+
return nil, ErrNotAuthorized
263
+
if resp.StatusCode != http.StatusOK {
264
+
return nil, fmt.Errorf("PDS returned status %d: %s", resp.StatusCode, string(body))
267
+
// Parse the listRecords response
268
+
var result struct {
270
+
URI string `json:"uri"`
271
+
CID string `json:"cid"`
273
+
Type string `json:"$type"`
275
+
URI string `json:"uri"`
276
+
CID string `json:"cid"`
278
+
Direction string `json:"direction"`
279
+
CreatedAt string `json:"createdAt"`
282
+
Cursor string `json:"cursor"`
285
+
if err := json.Unmarshal(body, &result); err != nil {
286
+
return nil, fmt.Errorf("failed to parse PDS response: %w", err)
289
+
// Search for the vote matching our subject in this page
290
+
for _, rec := range result.Records {
291
+
if rec.Value.Subject.URI == subjectURI {
292
+
// Extract rkey from the URI (at://did/collection/rkey)
293
+
parts := strings.Split(rec.URI, "/")
294
+
if len(parts) < 5 {
297
+
rkey := parts[len(parts)-1]
299
+
return &existingVote{
303
+
Direction: rec.Value.Direction,
308
+
// Check if there are more pages
309
+
if result.Cursor == "" {
310
+
break // No more pages
312
+
cursor = result.Cursor
315
+
// No vote found for this subject after checking all pages
319
+
// existingVote represents a vote record found on the PDS
320
+
type existingVote struct {
327
+
// deleteVoteRecord removes a vote record from the user's PDS
328
+
func (s *voteService) deleteVoteRecord(ctx context.Context, session *oauth.ClientSessionData, rkey string) error {
329
+
// Call com.atproto.repo.deleteRecord on the user's PDS
330
+
endpoint := fmt.Sprintf("%s/xrpc/com.atproto.repo.deleteRecord", strings.TrimSuffix(session.HostURL, "/"))
332
+
payload := map[string]interface{}{
333
+
"repo": session.AccountDID.String(),
334
+
"collection": "social.coves.feed.vote",
338
+
_, _, err := s.callPDSWithAuth(ctx, "POST", endpoint, payload, session.AccessToken)
342
+
// callPDSWithAuth makes an authenticated HTTP call to the PDS
343
+
// Returns URI and CID from the response (for create/update operations)
344
+
func (s *voteService) callPDSWithAuth(ctx context.Context, method, endpoint string, payload map[string]interface{}, accessToken string) (string, string, error) {
345
+
jsonData, err := json.Marshal(payload)
347
+
return "", "", fmt.Errorf("failed to marshal payload: %w", err)
350
+
req, err := http.NewRequestWithContext(ctx, method, endpoint, bytes.NewBuffer(jsonData))
352
+
return "", "", fmt.Errorf("failed to create request: %w", err)
354
+
req.Header.Set("Content-Type", "application/json")
356
+
// Add OAuth bearer token for authentication
357
+
if accessToken != "" {
358
+
req.Header.Set("Authorization", "Bearer "+accessToken)
361
+
// Set reasonable timeout for PDS operations
362
+
timeout := 10 * time.Second
363
+
if strings.Contains(endpoint, "createRecord") || strings.Contains(endpoint, "putRecord") {
364
+
timeout = 15 * time.Second // Slightly longer for write operations
367
+
client := &http.Client{Timeout: timeout}
368
+
resp, err := client.Do(req)
370
+
return "", "", fmt.Errorf("failed to call PDS: %w", err)
373
+
if closeErr := resp.Body.Close(); closeErr != nil {
374
+
s.logger.Warn("failed to close response body", "error", closeErr)
378
+
body, err := io.ReadAll(resp.Body)
380
+
return "", "", fmt.Errorf("failed to read response: %w", err)
383
+
// Handle auth errors - map to ErrNotAuthorized per lexicon
384
+
if resp.StatusCode == http.StatusUnauthorized || resp.StatusCode == http.StatusForbidden {
385
+
s.logger.Warn("PDS auth failure during write operation",
386
+
"status", resp.StatusCode,
387
+
"endpoint", endpoint)
388
+
return "", "", ErrNotAuthorized
391
+
if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusCreated {
392
+
return "", "", fmt.Errorf("PDS returned status %d: %s", resp.StatusCode, string(body))
395
+
// Parse response to extract URI and CID (for create/update operations)
396
+
var result struct {
397
+
URI string `json:"uri"`
398
+
CID string `json:"cid"`
400
+
if err := json.Unmarshal(body, &result); err != nil {
401
+
// For delete operations, there might not be a response body with URI/CID
402
+
if method == "POST" && strings.Contains(endpoint, "deleteRecord") {
405
+
return "", "", fmt.Errorf("failed to parse PDS response: %w", err)
408
+
return result.URI, result.CID, nil
bretton.dev