[bugfix] update fedi api to support multiple separate votes in same multiple choice poll (#3809)

internal/db/bundb/poll.go

@@ -380,8 +380,8 @@ func (p *pollDB) PutPollVote(ctx context.Context, vote *gtsmodel.PollVote) error
 				return err
 			}
 
-			// Increment poll votes for choices.
-			poll.IncrementVotes(vote.Choices)
+			// Increment vote choices and voters count.
+			poll.IncrementVotes(vote.Choices, true)
 
 			// Finally, update the poll entry.
 			_, err := tx.NewUpdate().
@@ -394,6 +394,17 @@ func (p *pollDB) PutPollVote(ctx context.Context, vote *gtsmodel.PollVote) error
 	})
 }
 
+func (p *pollDB) UpdatePollVote(ctx context.Context, vote *gtsmodel.PollVote, cols ...string) error {
+	return p.state.Caches.DB.PollVote.Store(vote, func() error {
+		_, err := p.db.NewUpdate().
+			Model(vote).
+			Column(cols...).
+			Where("? = ?", bun.Ident("id"), vote.ID).
+			Exec(ctx)
+		return err
+	})
+}
+
 func (p *pollDB) DeletePollVoteBy(ctx context.Context, pollID string, accountID string) error {
 	// Gather necessary fields from
 	// deleted for cache invaliation.
@@ -487,8 +498,8 @@ func updatePollCounts(ctx context.Context, tx bun.Tx, deleted *gtsmodel.PollVote
 		return err
 	}
 
-	// Decrement votes for these choices.
-	poll.DecrementVotes(deleted.Choices)
+	// Decrement vote choices and voters count.
+	poll.DecrementVotes(deleted.Choices, true)
 
 	// Finally, update the poll entry.
 	if _, err := tx.NewUpdate().

internal/db/poll.go

@@ -58,6 +58,9 @@ type Poll interface {
 	// PutPollVote puts the given PollVote in the database.
 	PutPollVote(ctx context.Context, vote *gtsmodel.PollVote) error
 
+	// UpdatePollVote updates the given poll vote in the database, using only given columns (else, all).
+	UpdatePollVote(ctx context.Context, vote *gtsmodel.PollVote, cols ...string) error
+
 	// DeletePollVoteBy deletes the PollVote in Poll with ID, by account ID, from the database.
 	DeletePollVoteBy(ctx context.Context, pollID string, accountID string) error
 

internal/federation/federatingdb/create.go

@@ -20,6 +20,7 @@
 import (
 	"context"
 	"errors"
+	"slices"
 
 	"github.com/superseriousbusiness/activity/streams/vocab"
 	"github.com/superseriousbusiness/gotosocial/internal/ap"
@@ -138,6 +139,10 @@ func (f *federatingDB) createPollOptionables(
 		// then re-used in each further iteration.
 		inReplyTo *gtsmodel.Status
 
+		// existing poll vote by requesting user
+		// in receiving user's poll, if it exists.
+		vote *gtsmodel.PollVote
+
 		// the resulting slices of Poll.Option
 		// choice indices passed into the new
 		// created PollVote object.
@@ -169,15 +174,17 @@ func (f *federatingDB) createPollOptionables(
 			case inReplyTo.PollID == "":
 				return gtserror.Newf("poll vote in status %s without poll", statusURI)
 
-			// We don't own the poll ...
-			case !*inReplyTo.Local:
-				return gtserror.Newf("poll vote in remote status %s", statusURI)
+			// Ensure poll isn't closed.
+			case inReplyTo.Poll.Closed():
+				return gtserror.Newf("poll vote in closed poll %s", statusURI)
+
+			// Ensure receiver actually owns poll.
+			case inReplyTo.AccountID != receiver.ID:
+				return gtserror.Newf("receiving account %s does not own poll %s", receiver.URI, statusURI)
 			}
 
-			// Check whether user has already vote in this poll.
-			// (we only check this for the first object, as multiple
-			// may be sent in response to a multiple-choice poll).
-			vote, err := f.state.DB.GetPollVoteBy(
+			// Try load any existing vote(s) by user.
+			vote, err = f.state.DB.GetPollVoteBy(
 				gtscontext.SetBarebones(ctx),
 				inReplyTo.PollID,
 				requester.ID,
@@ -185,11 +192,6 @@ func (f *federatingDB) createPollOptionables(
 			if err != nil && !errors.Is(err, db.ErrNoEntries) {
 				return gtserror.Newf("error getting status %s poll votes from database: %w", statusURI, err)
 			}
-
-			if vote != nil {
-				log.Warnf(ctx, "%s has already voted in poll %s", requester.URI, statusURI)
-				return nil // this is a useful warning for admins to report to us from logs
-			}
 		}
 
 		if statusURI != inReplyTo.URI {
@@ -210,7 +212,45 @@ func (f *federatingDB) createPollOptionables(
 		choices = append(choices, choice)
 	}
 
-	// Enqueue message to the fedi API worker with poll vote(s).
+	if vote != nil {
+		// Ensure this isn't a multiple vote
+		// by same account in the same poll.
+		if !*inReplyTo.Poll.Multiple {
+			log.Warnf(ctx, "%s has already voted in single-choice poll %s", requester.URI, inReplyTo.URI)
+			return nil // this is a useful warning for admins to report to us from logs
+		}
+
+		// Drop all existing votes from the new slice of choices.
+		choices = slices.DeleteFunc(choices, func(choice int) bool {
+			return slices.Contains(vote.Choices, choice)
+		})
+
+		// Update poll with new choices but *not* voters.
+		inReplyTo.Poll.IncrementVotes(choices, false)
+
+		// Append new vote choices to existing
+		// vote, then sort them by index size.
+		vote.Choices = append(vote.Choices, choices...)
+		slices.Sort(vote.Choices)
+
+		// Populate the poll vote,
+		// later used in fedi worker.
+		vote.Poll = inReplyTo.Poll
+		vote.Account = requester
+
+		// TODO: would it be useful to store an UpdatedAt field
+		// on poll votes? i'm not sure we'd actually use it...
+
+		// Enqueue an update event for poll vote to fedi API worker.
+		f.state.Workers.Federator.Queue.Push(&messages.FromFediAPI{
+			APActivityType: ap.ActivityUpdate,
+			APObjectType:   ap.ActivityQuestion,
+			GTSModel:       vote,
+			Receiving:      receiver,
+			Requesting:     requester,
+		})
+	} else {
+		// Create new poll vote and enqueue create to fedi API worker.
 		f.state.Workers.Federator.Queue.Push(&messages.FromFediAPI{
 			APActivityType: ap.ActivityCreate,
 			APObjectType:   ap.ActivityQuestion,
@@ -225,6 +265,7 @@ func (f *federatingDB) createPollOptionables(
 			Receiving:  receiver,
 			Requesting: requester,
 		})
+	}
 
 	return nil
 }

internal/gtsmodel/poll.go

@@ -60,8 +60,8 @@ func (p *Poll) Closed() bool {
 		time.Now().After(p.ClosedAt)
 }
 
-// IncrementVotes increments Poll vote and voter counts for given choices.
-func (p *Poll) IncrementVotes(choices []int) {
+// IncrementVotes increments Poll vote counts for given choices, and voters if 'isNew' is set.
+func (p *Poll) IncrementVotes(choices []int, isNew bool) {
 	if len(choices) == 0 {
 		return
 	}
@@ -69,11 +69,13 @@ func (p *Poll) IncrementVotes(choices []int) {
 	for _, choice := range choices {
 		p.Votes[choice]++
 	}
+	if isNew {
 		(*p.Voters)++
+	}
 }
 
-// DecrementVotes decrements Poll vote and voter counts for given choices.
-func (p *Poll) DecrementVotes(choices []int) {
+// DecrementVotes decrements Poll vote counts for given choices, and voters if 'withVoter' is set.
+func (p *Poll) DecrementVotes(choices []int, withVoter bool) {
 	if len(choices) == 0 {
 		return
 	}
@@ -83,7 +85,8 @@ func (p *Poll) DecrementVotes(choices []int) {
 			p.Votes[choice]--
 		}
 	}
-	if (*p.Voters) != 0 {
+	if (*p.Voters) != 0 &&
+		withVoter {
 		(*p.Voters)--
 	}
 }

internal/processing/polls/vote.go

@@ -93,7 +93,7 @@ func (p *Processor) PollVote(ctx context.Context, requester *gtsmodel.Account, p
 	// Before enqueuing it, increment the poll
 	// vote counts on the copy attached to the
 	// PollVote (that we also later return).
-	poll.IncrementVotes(choices)
+	poll.IncrementVotes(choices, true)
 
 	// Enqueue worker task to handle side-effects of user poll vote(s).
 	p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{

internal/processing/workers/fromfediapi.go

@@ -124,6 +124,10 @@ func (p *Processor) ProcessFromFediAPI(ctx context.Context, fMsg *messages.FromF
 		// UPDATE ACCOUNT
 		case ap.ActorPerson:
 			return p.fediAPI.UpdateAccount(ctx, fMsg)
+
+		// UPDATE QUESTION
+		case ap.ActivityQuestion:
+			return p.fediAPI.UpdatePollVote(ctx, fMsg)
 		}
 
 	// ACCEPT SOMETHING
@@ -355,7 +359,8 @@ func (p *fediAPI) CreatePollVote(ctx context.Context, fMsg *messages.FromFediAPI
 		return gtserror.Newf("cannot cast %T -> *gtsmodel.PollVote", fMsg.GTSModel)
 	}
 
-	// Insert the new poll vote in the database.
+	// Insert the new poll vote in the database, note this
+	// will handle updating votes on the poll model itself.
 	if err := p.state.DB.PutPollVote(ctx, vote); err != nil {
 		return gtserror.Newf("error inserting poll vote in db: %w", err)
 	}
@@ -376,9 +381,9 @@ func (p *fediAPI) CreatePollVote(ctx context.Context, fMsg *messages.FromFediAPI
 	status.Poll = vote.Poll
 
 	if *status.Local {
-		// Before federating it, increment the
-		// poll vote counts on our local copy.
-		status.Poll.IncrementVotes(vote.Choices)
+		// Before federating it, increment the poll vote
+		// and voter counts, *only on our local copy*.
+		status.Poll.IncrementVotes(vote.Choices, true)
 
 		// These were poll votes in a local status, we need to
 		// federate the updated status model with latest vote counts.
@@ -387,8 +392,43 @@ func (p *fediAPI) CreatePollVote(ctx context.Context, fMsg *messages.FromFediAPI
 		}
 	}
 
-	// Interaction counts changed on the source status, uncache from timelines.
-	p.surface.invalidateStatusFromTimelines(ctx, vote.Poll.StatusID)
+	// Interaction counts changed, uncache from timelines.
+	p.surface.invalidateStatusFromTimelines(ctx, status.ID)
+
+	return nil
+}
+
+func (p *fediAPI) UpdatePollVote(ctx context.Context, fMsg *messages.FromFediAPI) error {
+	// Cast poll vote type from the worker message.
+	vote, ok := fMsg.GTSModel.(*gtsmodel.PollVote)
+	if !ok {
+		return gtserror.Newf("cannot cast %T -> *gtsmodel.PollVote", fMsg.GTSModel)
+	}
+
+	// Update poll vote model (specifically only choices) in the database.
+	if err := p.state.DB.UpdatePollVote(ctx, vote, "choices"); err != nil {
+		return gtserror.Newf("error updating poll vote in db: %w", err)
+	}
+
+	// Update the vote counts on the poll model itself. These will have
+	// been updated by message pusher as we can't know which were new.
+	if err := p.state.DB.UpdatePoll(ctx, vote.Poll, "votes"); err != nil {
+		return gtserror.Newf("error updating poll in db: %w", err)
+	}
+
+	// Get the origin status.
+	status := vote.Poll.Status
+
+	if *status.Local {
+		// These were poll votes in a local status, we need to
+		// federate the updated status model with latest vote counts.
+		if err := p.federate.UpdateStatus(ctx, status); err != nil {
+			log.Errorf(ctx, "error federating status update: %v", err)
+		}
+	}
+
+	// Interaction counts changed, uncache from timelines.
+	p.surface.invalidateStatusFromTimelines(ctx, status.ID)
 
 	return nil
 }