17557469823

Committed 08 Sep 2025 04:25PM UTC coverage: 58.777% (-0.06%) from 58.841%

Build # 17557469823

Build Type

Pull #213

github

Committed by

altergui

Commit Message

fix

Pull Request Pull Request #213: api: standardize parameters ProcessID, CensusID, GroupID, JobID, UserID, BundleID

Run Details

254 of 345 new or added lines in 22 files covered. (73.62%)

19 existing lines in 7 files now uncovered.

5652 of 9616 relevant lines covered (58.78%)

32.01 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

79.28

/db/census_participant.go

package db

import (
        "context"
        "fmt"
        "time"

        "github.com/ethereum/go-ethereum/common"
        "github.com/vocdoni/saas-backend/internal"
        "go.mongodb.org/mongo-driver/bson"
        "go.mongodb.org/mongo-driver/mongo"
        "go.mongodb.org/mongo-driver/mongo/options"
        "go.vocdoni.io/dvote/log"
)

// validateCensusParticipant validates that a census participant can be created
// by checking that the census exists, the organization exists, and the member exists
func (ms *MongoStorage) validateCensusParticipant(participant *CensusParticipant) (common.Address, error) {
        // validate required fields
        if participant.ParticipantID.IsZero() || participant.CensusID.IsZero() {
                return common.Address{}, ErrInvalidData
        }

        // check that the published census exists
        census, err := ms.Census(participant.CensusID)
        if err != nil {
                return common.Address{}, fmt.Errorf("failed to get published census: %w", err)
        }

        // check that the org exists
        _, err = ms.Organization(census.OrgAddress)
        if err != nil {
                if err == ErrNotFound {
                        return common.Address{}, ErrInvalidData
                }
                return common.Address{}, fmt.Errorf("organization not found: %w", err)
        }

        // check that the member exists
        if _, err := ms.OrgMember(census.OrgAddress, participant.ParticipantID); err != nil {
                return common.Address{}, fmt.Errorf("failed to get org member: %w", err)
        }

        return census.OrgAddress, nil
}

// SetCensusParticipant creates or updates a census participant in the database.
// If the participant already exists (same participantID and censusID), it updates it.
// If it doesn't exist, it creates a new one.
func (ms *MongoStorage) SetCensusParticipant(participant *CensusParticipant) error {
        // Validate the participant
        _, err := ms.validateCensusParticipant(participant)
        if err != nil {
                return err
        }

        // prepare filter for upsert
        filter := bson.M{
                "participantID": participant.ParticipantID,
                "censusId":      participant.CensusID,
        }

        // set timestamps
        now := time.Now()
        participant.UpdatedAt = now
        if participant.CreatedAt.IsZero() {
                participant.CreatedAt = now
        }

        // create update document
        updateDoc := bson.M{
                "$set": participant,
        }

        // Perform database operation
        ms.keysLock.Lock()
        defer ms.keysLock.Unlock()

        ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)
        defer cancel()

        opts := options.Update().SetUpsert(true)
        if _, err := ms.censusParticipants.UpdateOne(ctx, filter, updateDoc, opts); err != nil {
                return fmt.Errorf("failed to set census participant: %w", err)
        }

        return nil
}

// CensusParticipant retrieves a census participant from the database based on
// participantID and censusID. Returns ErrNotFound if the participant doesn't exist.
func (ms *MongoStorage) CensusParticipant(censusID, id internal.ObjectID) (*CensusParticipant, error) {
        // create a context with a timeout
        ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)
        defer cancel()

        // validate input
        if id.IsZero() || censusID.IsZero() {
                return nil, ErrInvalidData
        }

        // prepare filter for find
        filter := bson.M{
                "participantID": id,
                "censusId":      censusID,
        }

        // find the participant
        participant := &CensusParticipant{}
        err := ms.censusParticipants.FindOne(ctx, filter).Decode(participant)
        if err != nil {
                if err == mongo.ErrNoDocuments {
                        return nil, ErrNotFound
                }
                return nil, fmt.Errorf("failed to get census participant: %w", err)
        }

        return participant, nil
}

// CensusParticipantByMemberNumber retrieves a census participant from the database based on
// memberNumber and censusID. Returns ErrNotFound if the participant doesn't exist.
func (ms *MongoStorage) CensusParticipantByMemberNumber(
        censusID internal.ObjectID,
        memberNumber string,
        orgAddress common.Address,
) (*CensusParticipant, error) {
        // create a context with a timeout
        ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)
        defer cancel()

        // validate input
        if len(memberNumber) == 0 || censusID.IsZero() {
                return nil, ErrInvalidData
        }

        orgMember, err := ms.OrgMemberByMemberNumber(orgAddress, memberNumber)
        if err != nil {
                if err == mongo.ErrNoDocuments || err == ErrNotFound {
                        return nil, ErrNotFound
                }
                return nil, fmt.Errorf("failed to get org member: %w", err)
        }

        // prepare filter for find
        filter := bson.M{
                "participantID": orgMember.ID,
                "censusId":      censusID,
        }

        // find the participant
        participant := &CensusParticipant{}
        err = ms.censusParticipants.FindOne(ctx, filter).Decode(participant)
        if err != nil {
                if err == mongo.ErrNoDocuments {
                        return nil, ErrNotFound
                }
                return nil, fmt.Errorf("failed to get census participant: %w", err)
        }

        return participant, nil
}

// CensusParticipantByLoginHash retrieves a census participant from the database based on
// the login data hash and censusID. Returns ErrNotFound if the participant doesn't exist.
// TODO add the index
func (ms *MongoStorage) CensusParticipantByLoginHash(
        censusID internal.ObjectID,
        loginHash []byte,
) (*CensusParticipant, error) {
        // create a context with a timeout
        ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)
        defer cancel()

        // validate input
        if len(loginHash) == 0 || censusID.IsZero() {
                return nil, ErrInvalidData
        }

        // prepare filter for find
        filter := bson.M{
                "loginHash": loginHash,
                "censusId":  censusID,
        }

        // find the participant
        participant := &CensusParticipant{}
        err := ms.censusParticipants.FindOne(ctx, filter).Decode(participant)
        if err != nil {
                if err == mongo.ErrNoDocuments {
                        return nil, ErrNotFound
                }
                return nil, fmt.Errorf("failed to get census participant: %w", err)
        }

        return participant, nil
}

// DelCensusParticipant removes a census participant from the database.
// Returns nil if the participant was successfully deleted or didn't exist.
func (ms *MongoStorage) DelCensusParticipant(censusID, participantID internal.ObjectID) error {
        ms.keysLock.Lock()
        defer ms.keysLock.Unlock()
        // create a context with a timeout
        ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)
        defer cancel()

        // validate input
        if participantID.IsZero() || censusID.IsZero() {
                return ErrInvalidData
        }

        // prepare filter for upsert
        filter := bson.M{
                "participantID": participantID,
                "censusId":      censusID,
        }

        // delete the participant
        _, err := ms.censusParticipants.DeleteOne(ctx, filter)
        if err != nil {
                return fmt.Errorf("failed to delete census participant: %w", err)
        }

        return nil
}

// BulkCensusParticipantStatus is returned by SetBylkCensusParticipant to provide the output.
type BulkCensusParticipantStatus struct {
        Progress int `json:"progress"`
        Total    int `json:"total"`
        Added    int `json:"added"`
}

// createCensusParticipantBulkOperations creates the bulk write operations for members and participants
func createCensusParticipantBulkOperations(
        orgMembers []*OrgMember,
        org *Organization,
        censusID internal.ObjectID,
        salt string,
        currentTime time.Time,
) (orgMembersOps []mongo.WriteModel, censusParticipantOps []mongo.WriteModel) {
        var bulkOrgMembersOps []mongo.WriteModel
        var bulkCensusParticipantsOps []mongo.WriteModel

        for _, m := range orgMembers {
                // Prepare the member
                orgMember, _ := prepareOrgMember(org, m, salt, currentTime)
                // TODO: handle prepareOrgMember []error, pass them back to client

                // Create member filter and update document
                memberFilter := bson.M{
                        "_id":        orgMember.ID,
                        "orgAddress": orgMember.OrgAddress,
                }

                updateOrgMembersDoc, err := dynamicUpdateDocument(orgMember, nil)
                if err != nil {
                        log.Warnw("failed to create update document for member",
                                "error", err, "ID", orgMember.ID)
                        continue // Skip this member but continue with others
                }

                // Create member upsert model
                upsertOrgMembersModel := mongo.NewUpdateOneModel().
                        SetFilter(memberFilter).
                        SetUpdate(updateOrgMembersDoc).
                        SetUpsert(true)
                bulkOrgMembersOps = append(bulkOrgMembersOps, upsertOrgMembersModel)

                // Create participant filter and document
                censusParticipantsFilter := bson.M{
                        "participantID": orgMember.ID,
                        "censusId":      censusID,
                }

                // Create document for $set operation (without CreatedAt)
                participantDoc := &CensusParticipant{
                        ParticipantID: orgMember.ID,
                        CensusID:      censusID,
                        UpdatedAt:     currentTime,
                }

                // Create participant update document
                updateParticipantDoc, err := dynamicUpdateDocument(participantDoc, nil)
                if err != nil {
                        log.Warnw("failed to create update document for participant",
                                "error", err, "participantID", orgMember.ID)
                        continue
                }

                // Extract the $set part from the update document with type checking
                setDoc, ok := updateParticipantDoc["$set"].(bson.M)
                if !ok {
                        log.Warnw("failed to extract $set document for participant",
                                "error", "invalid $set type", "participantID", orgMember.ID)
                        continue
                }

                // Create combined update document with both $set and $setOnInsert
                combinedUpdateDoc := bson.M{
                        "$set": setDoc,
                        "$setOnInsert": bson.M{
                                "createdAt": currentTime,
                        },
                }

                // Create participant upsert model
                upsertCensusParticipantsModel := mongo.NewUpdateOneModel().
                        SetFilter(censusParticipantsFilter).
                        SetUpdate(combinedUpdateDoc).
                        SetUpsert(true)
                bulkCensusParticipantsOps = append(bulkCensusParticipantsOps, upsertCensusParticipantsModel)
        }

        return bulkOrgMembersOps, bulkCensusParticipantsOps
}

// processBatch processes a batch of members and returns the number added
func (ms *MongoStorage) processBatch(
        bulkOrgMembersOps []mongo.WriteModel,
        bulkCensusParticipantOps []mongo.WriteModel,
) int {
        if len(bulkOrgMembersOps) == 0 {
                return 0
        }

        // Only lock the mutex during the actual database operations
        ms.keysLock.Lock()
        defer ms.keysLock.Unlock()

        // Create a new context for the batch
        batchCtx, batchCancel := context.WithTimeout(context.Background(), batchTimeout)
        defer batchCancel()

        // Execute the bulk write operations for org members
        _, err := ms.orgMembers.BulkWrite(batchCtx, bulkOrgMembersOps)
        if err != nil {
                log.Warnw("failed to perform bulk operation on members", "error", err)
                return 0
        }

        // Execute the bulk write operations for census participants
        _, err = ms.censusParticipants.BulkWrite(batchCtx, bulkCensusParticipantOps)
        if err != nil {
                log.Warnw("failed to perform bulk operation on participants", "error", err)
                return 0
        }

        return len(bulkOrgMembersOps)
}

// startProgressReporter starts a goroutine that reports progress periodically
func startProgressReporter(
        ctx context.Context,
        progressChan chan<- *BulkCensusParticipantStatus,
        totalOrgMembers int,
        processedOrgMembers *int,
        addedOrgMembers *int,
) {
        ticker := time.NewTicker(10 * time.Second)
        defer ticker.Stop()

        for {
                select {
                case <-ticker.C:
                        // Calculate and send progress percentage
                        if totalOrgMembers > 0 {
                                progress := (*processedOrgMembers * 100) / totalOrgMembers
                                progressChan <- &BulkCensusParticipantStatus{
                                        Progress: progress,
                                        Total:    totalOrgMembers,
                                        Added:    *addedOrgMembers,
                                }
                        }
                case <-ctx.Done():
                        return
                }
        }
}

// validateBulkCensusParticipant validates the input parameters for bulk census participant
// and returns the census if valid
func (ms *MongoStorage) validateBulkCensusParticipant(
        censusID internal.ObjectID,
        orgMembersSize int,
) (*Census, error) {
        // Early returns for invalid input
        if orgMembersSize == 0 {
                return nil, nil // Not an error, just no work to do
        }
        if censusID.IsZero() {
                return nil, ErrInvalidData
        }

        // Validate census and organization
        census, err := ms.Census(censusID)
        if err != nil {
                return nil, fmt.Errorf("failed to get published census: %w", err)
        }

        if _, err := ms.Organization(census.OrgAddress); err != nil {
                return nil, err
        }

        return census, nil
}

// processBatches processes members in batches and sends progress updates
func (ms *MongoStorage) processBatches(
        orgMembers []*OrgMember,
        org *Organization,
        census *Census,
        salt string,
        progressChan chan<- *BulkCensusParticipantStatus,
) {
        defer close(progressChan)

        // Process members in batches of 200
        batchSize := 200
        totalOrgMembers := len(orgMembers)
        processedOrgMembers := 0
        addedOrgMembers := 0
        currentTime := time.Now()

        // Send initial progress
        progressChan <- &BulkCensusParticipantStatus{
                Progress: 0,
                Total:    totalOrgMembers,
                Added:    addedOrgMembers,
        }

        // Create a context for the entire operation
        ctx, cancel := context.WithCancel(context.Background())
        defer cancel()

        // Start progress reporter in a separate goroutine
        go startProgressReporter(ctx, progressChan, totalOrgMembers, &processedOrgMembers, &addedOrgMembers)

        // Process members in batches
        for i := 0; i < totalOrgMembers; i += batchSize {
                // Calculate end index for current batch
                end := i + batchSize
                if end > totalOrgMembers {
                        end = totalOrgMembers
                }

                // Create bulk operations for this batch
                bulkOrgMembersOps, bulkCensusParticipantOps := createCensusParticipantBulkOperations(
                        orgMembers[i:end],
                        org,
                        census.ID,
                        salt,
                        currentTime,
                )

                // Process the batch and get number of added members
                added := ms.processBatch(bulkOrgMembersOps, bulkCensusParticipantOps)
                addedOrgMembers += added

                // Update processed count
                processedOrgMembers += (end - i)
        }

        // Send final progress (100%)
        progressChan <- &BulkCensusParticipantStatus{
                Progress: 100,
                Total:    totalOrgMembers,
                Added:    addedOrgMembers,
        }
}

// SetBulkCensusOrgMemberParticipant creates or updates an org member and a census participant in the database.
// If the participant already exists (same participantID and censusID), it updates it.
// If it doesn't exist, it creates a new one.
// Processes members in batches of 200 entries.
// Returns a channel that sends the percentage of members processed every 10 seconds.
// This function must be called in a goroutine.
func (ms *MongoStorage) SetBulkCensusOrgMemberParticipant(
        org *Organization, salt string, censusID internal.ObjectID, orgMembers []*OrgMember,
) (chan *BulkCensusParticipantStatus, error) {
        progressChan := make(chan *BulkCensusParticipantStatus, 10)

        // Validate input parameters
        census, err := ms.validateBulkCensusParticipant(censusID, len(orgMembers))
        if err != nil {
                close(progressChan)
                return progressChan, err
        }

        // If no members, return empty channel
        if census == nil {
                close(progressChan)
                return progressChan, nil
        }

        // Start processing in a goroutine
        go ms.processBatches(orgMembers, org, census, salt, progressChan)

        return progressChan, nil
}

func (ms *MongoStorage) setBulkCensusParticipant(
        ctx context.Context, censusID internal.ObjectID, groupID internal.ObjectID, orgAddress common.Address,
        authFields OrgMemberAuthFields, twoFaFields OrgMemberTwoFaFields,
) (int64, error) {
        _, members, err := ms.ListOrganizationMemberGroup(groupID, orgAddress, 0, 0)
        if err != nil {
                return 0, fmt.Errorf("error retrieving group members: %w", err)
        }
        if len(members) == 0 {
                return 0, nil // nothing to do
        }

        // prepare filter for upsert
        currentTime := time.Now()

        docs := make([]mongo.WriteModel, 0, len(members))
        for _, member := range members {
                // Create participant filter and document
                censusParticipantsFilter := bson.M{
                        "participantID": member.ID,
                        "censusId":      censusID,
                }
                participantDoc := &CensusParticipant{
                        ParticipantID: member.ID,
                        LoginHash:     HashAuthTwoFaFields(*member, authFields, twoFaFields),
                        CensusID:      censusID,
                        UpdatedAt:     currentTime,
                }

                // Create participant update document
                updateParticipantDoc, err := dynamicUpdateDocument(participantDoc, nil)
                if err != nil {
                        log.Warnw("failed to create update document for participant",
                                "error", err, "participantID", member.ID)
                        continue
                }

                // Extract the $set part from the update document with type checking
                setDoc, ok := updateParticipantDoc["$set"].(bson.M)
                if !ok {
                        log.Warnw("failed to extract $set document for participant",
                                "error", "invalid $set type", "participantID", member.ID)
                        continue
                }

                // Create combined update document with both $set and $setOnInsert
                combinedUpdateDoc := bson.M{
                        "$set": setDoc,
                        "$setOnInsert": bson.M{
                                "createdAt": currentTime,
                        },
                }

                // Create participant upsert model
                upsertCensusParticipantsModel := mongo.NewUpdateOneModel().
                        SetFilter(censusParticipantsFilter).
                        SetUpdate(combinedUpdateDoc).
                        SetUpsert(true)
                docs = append(docs, upsertCensusParticipantsModel)
        }
        // Unordered makes it continue on errors (e.g., one dup)
        bulkOpts := options.BulkWrite().SetOrdered(false)

        results, err := ms.censusParticipants.BulkWrite(ctx, docs, bulkOpts)
        return results.UpsertedCount, err
}

// CensusParticipants retrieves all the census participants for a given census.
func (ms *MongoStorage) CensusParticipants(censusID internal.ObjectID) ([]CensusParticipant, error) {
        // create a context with a timeout
        ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)
        defer cancel()

        // validate input
        if censusID.IsZero() {
                return nil, ErrInvalidData
        }

        // prepare filter for upsert
        filter := bson.M{
                "censusId": censusID,
        }

        // find the participant
        cursor, err := ms.censusParticipants.Find(ctx, filter)
        if err != nil {
                return nil, fmt.Errorf("failed to get census participants: %w", err)
        }
        defer func() {
                if err := cursor.Close(ctx); err != nil {
                        log.Warnw("error closing cursor", "error", err)
                }
        }()
        var participants []CensusParticipant
        if err := cursor.All(ctx, &participants); err != nil {
                return nil, fmt.Errorf("failed to get census participants: %w", err)
        }

        return participants, nil
}

1	package db
2
3	import (
4	"context"
5	"fmt"
6	"time"
7
8	"github.com/ethereum/go-ethereum/common"
9	"github.com/vocdoni/saas-backend/internal"
10	"go.mongodb.org/mongo-driver/bson"
11	"go.mongodb.org/mongo-driver/mongo"
12	"go.mongodb.org/mongo-driver/mongo/options"
13	"go.vocdoni.io/dvote/log"
14	)
15
16	// validateCensusParticipant validates that a census participant can be created
17	// by checking that the census exists, the organization exists, and the member exists
18	func (ms MongoStorage) validateCensusParticipant(participant CensusParticipant) (common.Address, error) {	9✔
19	// validate required fields	9✔
20	if participant.ParticipantID.IsZero() \|\| participant.CensusID.IsZero() {	11✔
21	return common.Address{}, ErrInvalidData	2✔
22	}	2✔
23
24	// check that the published census exists
25	census, err := ms.Census(participant.CensusID)	7✔
26	if err != nil {	8✔
27	return common.Address{}, fmt.Errorf("failed to get published census: %w", err)	1✔
28	}	1✔
29
30	// check that the org exists
31	_, err = ms.Organization(census.OrgAddress)	6✔
32	if err != nil {	6✔
33	if err == ErrNotFound {	×
34	return common.Address{}, ErrInvalidData	×
35	}	×
36	return common.Address{}, fmt.Errorf("organization not found: %w", err)	×
37	}
38
39	// check that the member exists
40	if _, err := ms.OrgMember(census.OrgAddress, participant.ParticipantID); err != nil {	7✔
41	return common.Address{}, fmt.Errorf("failed to get org member: %w", err)	1✔
42	}	1✔
43
44	return census.OrgAddress, nil	5✔
45	}
46
47	// SetCensusParticipant creates or updates a census participant in the database.
48	// If the participant already exists (same participantID and censusID), it updates it.
49	// If it doesn't exist, it creates a new one.
50	func (ms MongoStorage) SetCensusParticipant(participant CensusParticipant) error {	9✔
51	// Validate the participant	9✔
52	_, err := ms.validateCensusParticipant(participant)	9✔
53	if err != nil {	13✔
54	return err	4✔
55	}	4✔
56
57	// prepare filter for upsert
58	filter := bson.M{	5✔
59	"participantID": participant.ParticipantID,	5✔
60	"censusId": participant.CensusID,	5✔
61	}	5✔
62		5✔
63	// set timestamps	5✔
64	now := time.Now()	5✔
65	participant.UpdatedAt = now	5✔
66	if participant.CreatedAt.IsZero() {	9✔
67	participant.CreatedAt = now	4✔
68	}	4✔
69
70	// create update document
71	updateDoc := bson.M{	5✔
72	"$set": participant,	5✔
73	}	5✔
74		5✔
75	// Perform database operation	5✔
76	ms.keysLock.Lock()	5✔
77	defer ms.keysLock.Unlock()	5✔
78		5✔
79	ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)	5✔
80	defer cancel()	5✔
81		5✔
82	opts := options.Update().SetUpsert(true)	5✔
83	if _, err := ms.censusParticipants.UpdateOne(ctx, filter, updateDoc, opts); err != nil {	5✔
84	return fmt.Errorf("failed to set census participant: %w", err)	×
85	}	×
86
87	return nil	5✔
88	}
89
90	// CensusParticipant retrieves a census participant from the database based on
91	// participantID and censusID. Returns ErrNotFound if the participant doesn't exist.
92	func (ms MongoStorage) CensusParticipant(censusID, id internal.ObjectID) (CensusParticipant, error) {	11✔
93	// create a context with a timeout	11✔
94	ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)	11✔
95	defer cancel()	11✔
96		11✔
97	// validate input	11✔
98	if id.IsZero() \|\| censusID.IsZero() {	13✔
99	return nil, ErrInvalidData	2✔
100	}	2✔
101
102	// prepare filter for find
103	filter := bson.M{	9✔
104	"participantID": id,	9✔
105	"censusId": censusID,	9✔
106	}	9✔
107		9✔
108	// find the participant	9✔
109	participant := &CensusParticipant{}	9✔
110	err := ms.censusParticipants.FindOne(ctx, filter).Decode(participant)	9✔
111	if err != nil {	11✔
112	if err == mongo.ErrNoDocuments {	4✔
113	return nil, ErrNotFound	2✔
114	}	2✔
115	return nil, fmt.Errorf("failed to get census participant: %w", err)	×
116	}
117
118	return participant, nil	7✔
119	}
120
121	// CensusParticipantByMemberNumber retrieves a census participant from the database based on
122	// memberNumber and censusID. Returns ErrNotFound if the participant doesn't exist.
123	func (ms *MongoStorage) CensusParticipantByMemberNumber(
124	censusID internal.ObjectID,
125	memberNumber string,
126	orgAddress common.Address,
127	) (*CensusParticipant, error) {	×
128	// create a context with a timeout	×
129	ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)	×
130	defer cancel()	×
131		×
132	// validate input	×
NEW 133	if len(memberNumber) == 0 \|\| censusID.IsZero() {	×
134	return nil, ErrInvalidData	×
135	}	×
136
137	orgMember, err := ms.OrgMemberByMemberNumber(orgAddress, memberNumber)	×
138	if err != nil {	×
139	if err == mongo.ErrNoDocuments \|\| err == ErrNotFound {	×
140	return nil, ErrNotFound	×
141	}	×
142	return nil, fmt.Errorf("failed to get org member: %w", err)	×
143	}
144
145	// prepare filter for find
146	filter := bson.M{	×
NEW 147	"participantID": orgMember.ID,	×
148	"censusId": censusID,	×
149	}	×
150		×
151	// find the participant	×
152	participant := &CensusParticipant{}	×
153	err = ms.censusParticipants.FindOne(ctx, filter).Decode(participant)	×
154	if err != nil {	×
155	if err == mongo.ErrNoDocuments {	×
156	return nil, ErrNotFound	×
157	}	×
158	return nil, fmt.Errorf("failed to get census participant: %w", err)	×
159	}
160
161	return participant, nil	×
162	}
163
164	// CensusParticipantByLoginHash retrieves a census participant from the database based on
165	// the login data hash and censusID. Returns ErrNotFound if the participant doesn't exist.
166	// TODO add the index
167	func (ms *MongoStorage) CensusParticipantByLoginHash(
168	censusID internal.ObjectID,
169	loginHash []byte,
170	) (*CensusParticipant, error) {	23✔
171	// create a context with a timeout	23✔
172	ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)	23✔
173	defer cancel()	23✔
174		23✔
175	// validate input	23✔
176	if len(loginHash) == 0 \|\| censusID.IsZero() {	25✔
177	return nil, ErrInvalidData	2✔
178	}	2✔
179
180	// prepare filter for find
181	filter := bson.M{	21✔
182	"loginHash": loginHash,	21✔
183	"censusId": censusID,	21✔
184	}	21✔
185		21✔
186	// find the participant	21✔
187	participant := &CensusParticipant{}	21✔
188	err := ms.censusParticipants.FindOne(ctx, filter).Decode(participant)	21✔
189	if err != nil {	28✔
190	if err == mongo.ErrNoDocuments {	14✔
191	return nil, ErrNotFound	7✔
192	}	7✔
193	return nil, fmt.Errorf("failed to get census participant: %w", err)	×
194	}
195
196	return participant, nil	14✔
197	}
198
199	// DelCensusParticipant removes a census participant from the database.
200	// Returns nil if the participant was successfully deleted or didn't exist.
201	func (ms *MongoStorage) DelCensusParticipant(censusID, participantID internal.ObjectID) error {	4✔
202	ms.keysLock.Lock()	4✔
203	defer ms.keysLock.Unlock()	4✔
204	// create a context with a timeout	4✔
205	ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)	4✔
206	defer cancel()	4✔
207		4✔
208	// validate input	4✔
209	if participantID.IsZero() \|\| censusID.IsZero() {	6✔
210	return ErrInvalidData	2✔
211	}	2✔
212
213	// prepare filter for upsert
214	filter := bson.M{	2✔
215	"participantID": participantID,	2✔
216	"censusId": censusID,	2✔
217	}	2✔
218		2✔
219	// delete the participant	2✔
220	_, err := ms.censusParticipants.DeleteOne(ctx, filter)	2✔
221	if err != nil {	2✔
222	return fmt.Errorf("failed to delete census participant: %w", err)	×
223	}	×
224
225	return nil	2✔
226	}
227
228	// BulkCensusParticipantStatus is returned by SetBylkCensusParticipant to provide the output.
229	type BulkCensusParticipantStatus struct {
230	Progress int `json:"progress"`
231	Total int `json:"total"`
232	Added int `json:"added"`
233	}
234
235	// createCensusParticipantBulkOperations creates the bulk write operations for members and participants
236	func createCensusParticipantBulkOperations(
237	orgMembers []*OrgMember,
238	org *Organization,
239	censusID internal.ObjectID,
240	salt string,
241	currentTime time.Time,
242	) (orgMembersOps []mongo.WriteModel, censusParticipantOps []mongo.WriteModel) {	8✔
243	var bulkOrgMembersOps []mongo.WriteModel	8✔
244	var bulkCensusParticipantsOps []mongo.WriteModel	8✔
245		8✔
246	for _, m := range orgMembers {	24✔
247	// Prepare the member	16✔
248	orgMember, _ := prepareOrgMember(org, m, salt, currentTime)	16✔
249	// TODO: handle prepareOrgMember []error, pass them back to client	16✔
250		16✔
251	// Create member filter and update document	16✔
252	memberFilter := bson.M{	16✔
253	"_id": orgMember.ID,	16✔
254	"orgAddress": orgMember.OrgAddress,	16✔
255	}	16✔
256		16✔
257	updateOrgMembersDoc, err := dynamicUpdateDocument(orgMember, nil)	16✔
258	if err != nil {	16✔
259	log.Warnw("failed to create update document for member",	×
260	"error", err, "ID", orgMember.ID)	×
261	continue // Skip this member but continue with others	×
262	}
263
264	// Create member upsert model
265	upsertOrgMembersModel := mongo.NewUpdateOneModel().	16✔
266	SetFilter(memberFilter).	16✔
267	SetUpdate(updateOrgMembersDoc).	16✔
268	SetUpsert(true)	16✔
269	bulkOrgMembersOps = append(bulkOrgMembersOps, upsertOrgMembersModel)	16✔
270		16✔
271	// Create participant filter and document	16✔
272	censusParticipantsFilter := bson.M{	16✔
273	"participantID": orgMember.ID,	16✔
274	"censusId": censusID,	16✔
275	}	16✔
276		16✔
277	// Create document for $set operation (without CreatedAt)	16✔
278	participantDoc := &CensusParticipant{	16✔
279	ParticipantID: orgMember.ID,	16✔
280	CensusID: censusID,	16✔
281	UpdatedAt: currentTime,	16✔
282	}	16✔
283		16✔
284	// Create participant update document	16✔
285	updateParticipantDoc, err := dynamicUpdateDocument(participantDoc, nil)	16✔
286	if err != nil {	16✔
287	log.Warnw("failed to create update document for participant",	×
NEW 288	"error", err, "participantID", orgMember.ID)	×
289	continue	×
290	}
291
292	// Extract the $set part from the update document with type checking
293	setDoc, ok := updateParticipantDoc["$set"].(bson.M)	16✔
294	if !ok {	16✔
295	log.Warnw("failed to extract $set document for participant",	×
NEW 296	"error", "invalid $set type", "participantID", orgMember.ID)	×
297	continue	×
298	}
299
300	// Create combined update document with both $set and $setOnInsert
301	combinedUpdateDoc := bson.M{	16✔
302	"$set": setDoc,	16✔
303	"$setOnInsert": bson.M{	16✔
304	"createdAt": currentTime,	16✔
305	},	16✔
306	}	16✔
307		16✔
308	// Create participant upsert model	16✔
309	upsertCensusParticipantsModel := mongo.NewUpdateOneModel().	16✔
310	SetFilter(censusParticipantsFilter).	16✔
311	SetUpdate(combinedUpdateDoc).	16✔
312	SetUpsert(true)	16✔
313	bulkCensusParticipantsOps = append(bulkCensusParticipantsOps, upsertCensusParticipantsModel)	16✔
314	}
315
316	return bulkOrgMembersOps, bulkCensusParticipantsOps	8✔
317	}
318
319	// processBatch processes a batch of members and returns the number added
320	func (ms *MongoStorage) processBatch(
321	bulkOrgMembersOps []mongo.WriteModel,
322	bulkCensusParticipantOps []mongo.WriteModel,
323	) int {	8✔
324	if len(bulkOrgMembersOps) == 0 {	8✔
325	return 0	×
326	}	×
327
328	// Only lock the mutex during the actual database operations
329	ms.keysLock.Lock()	8✔
330	defer ms.keysLock.Unlock()	8✔
331		8✔
332	// Create a new context for the batch	8✔
333	batchCtx, batchCancel := context.WithTimeout(context.Background(), batchTimeout)	8✔
334	defer batchCancel()	8✔
335		8✔
336	// Execute the bulk write operations for org members	8✔
337	_, err := ms.orgMembers.BulkWrite(batchCtx, bulkOrgMembersOps)	8✔
338	if err != nil {	8✔
339	log.Warnw("failed to perform bulk operation on members", "error", err)	×
340	return 0	×
341	}	×
342
343	// Execute the bulk write operations for census participants
344	_, err = ms.censusParticipants.BulkWrite(batchCtx, bulkCensusParticipantOps)	8✔
345	if err != nil {	8✔
346	log.Warnw("failed to perform bulk operation on participants", "error", err)	×
347	return 0	×
348	}	×
349
350	return len(bulkOrgMembersOps)	8✔
351	}
352
353	// startProgressReporter starts a goroutine that reports progress periodically
354	func startProgressReporter(
355	ctx context.Context,
356	progressChan chan<- *BulkCensusParticipantStatus,
357	totalOrgMembers int,
358	processedOrgMembers *int,
359	addedOrgMembers *int,
360	) {	6✔
361	ticker := time.NewTicker(10 * time.Second)	6✔
362	defer ticker.Stop()	6✔
363		6✔
364	for {	12✔
365	select {	6✔
366	case <-ticker.C:	×
367	// Calculate and send progress percentage	×
368	if totalOrgMembers > 0 {	×
369	progress := (processedOrgMembers 100) / totalOrgMembers	×
370	progressChan <- &BulkCensusParticipantStatus{	×
371	Progress: progress,	×
372	Total: totalOrgMembers,	×
373	Added: *addedOrgMembers,	×
374	}	×
375	}	×
376	case <-ctx.Done():	6✔
377	return	6✔
378	}
379	}
380	}
381
382	// validateBulkCensusParticipant validates the input parameters for bulk census participant
383	// and returns the census if valid
384	func (ms *MongoStorage) validateBulkCensusParticipant(
385	censusID internal.ObjectID,
386	orgMembersSize int,
387	) (*Census, error) {	9✔
388	// Early returns for invalid input	9✔
389	if orgMembersSize == 0 {	10✔
390	return nil, nil // Not an error, just no work to do	1✔
391	}	1✔
392	if censusID.IsZero() {	9✔
393	return nil, ErrInvalidData	1✔
394	}	1✔
395
396	// Validate census and organization
397	census, err := ms.Census(censusID)	7✔
398	if err != nil {	8✔
399	return nil, fmt.Errorf("failed to get published census: %w", err)	1✔
400	}	1✔
401
402	if _, err := ms.Organization(census.OrgAddress); err != nil {	6✔
403	return nil, err	×
404	}	×
405
406	return census, nil	6✔
407	}
408
409	// processBatches processes members in batches and sends progress updates
410	func (ms *MongoStorage) processBatches(
411	orgMembers []*OrgMember,
412	org *Organization,
413	census *Census,
414	salt string,
415	progressChan chan<- *BulkCensusParticipantStatus,
416	) {	6✔
417	defer close(progressChan)	6✔
418		6✔
419	// Process members in batches of 200	6✔
420	batchSize := 200	6✔
421	totalOrgMembers := len(orgMembers)	6✔
422	processedOrgMembers := 0	6✔
423	addedOrgMembers := 0	6✔
424	currentTime := time.Now()	6✔
425		6✔
426	// Send initial progress	6✔
427	progressChan <- &BulkCensusParticipantStatus{	6✔
428	Progress: 0,	6✔
429	Total: totalOrgMembers,	6✔
430	Added: addedOrgMembers,	6✔
431	}	6✔
432		6✔
433	// Create a context for the entire operation	6✔
434	ctx, cancel := context.WithCancel(context.Background())	6✔
435	defer cancel()	6✔
436		6✔
437	// Start progress reporter in a separate goroutine	6✔
438	go startProgressReporter(ctx, progressChan, totalOrgMembers, &processedOrgMembers, &addedOrgMembers)	6✔
439		6✔
440	// Process members in batches	6✔
441	for i := 0; i < totalOrgMembers; i += batchSize {	12✔
442	// Calculate end index for current batch	6✔
443	end := i + batchSize	6✔
444	if end > totalOrgMembers {	12✔
445	end = totalOrgMembers	6✔
446	}	6✔
447
448	// Create bulk operations for this batch
449	bulkOrgMembersOps, bulkCensusParticipantOps := createCensusParticipantBulkOperations(	6✔
450	orgMembers[i:end],	6✔
451	org,	6✔
452	census.ID,	6✔
453	salt,	6✔
454	currentTime,	6✔
455	)	6✔
456		6✔
457	// Process the batch and get number of added members	6✔
458	added := ms.processBatch(bulkOrgMembersOps, bulkCensusParticipantOps)	6✔
459	addedOrgMembers += added	6✔
460		6✔
461	// Update processed count	6✔
462	processedOrgMembers += (end - i)	6✔
463	}
464
465	// Send final progress (100%)
466	progressChan <- &BulkCensusParticipantStatus{	6✔
467	Progress: 100,	6✔
468	Total: totalOrgMembers,	6✔
469	Added: addedOrgMembers,	6✔
470	}	6✔
471	}
472
473	// SetBulkCensusOrgMemberParticipant creates or updates an org member and a census participant in the database.
474	// If the participant already exists (same participantID and censusID), it updates it.
475	// If it doesn't exist, it creates a new one.
476	// Processes members in batches of 200 entries.
477	// Returns a channel that sends the percentage of members processed every 10 seconds.
478	// This function must be called in a goroutine.
479	func (ms *MongoStorage) SetBulkCensusOrgMemberParticipant(
480	org Organization, salt string, censusID internal.ObjectID, orgMembers []OrgMember,
481	) (chan *BulkCensusParticipantStatus, error) {	9✔
482	progressChan := make(chan *BulkCensusParticipantStatus, 10)	9✔
483		9✔
484	// Validate input parameters	9✔
485	census, err := ms.validateBulkCensusParticipant(censusID, len(orgMembers))	9✔
486	if err != nil {	11✔
487	close(progressChan)	2✔
488	return progressChan, err	2✔
489	}	2✔
490
491	// If no members, return empty channel
492	if census == nil {	8✔
493	close(progressChan)	1✔
494	return progressChan, nil	1✔
495	}	1✔
496
497	// Start processing in a goroutine
498	go ms.processBatches(orgMembers, org, census, salt, progressChan)	6✔
499		6✔
500	return progressChan, nil	6✔
501	}
502
503	func (ms *MongoStorage) setBulkCensusParticipant(
504	ctx context.Context, censusID internal.ObjectID, groupID internal.ObjectID, orgAddress common.Address,
505	authFields OrgMemberAuthFields, twoFaFields OrgMemberTwoFaFields,
506	) (int64, error) {	12✔
507	_, members, err := ms.ListOrganizationMemberGroup(groupID, orgAddress, 0, 0)	12✔
508	if err != nil {	12✔
509	return 0, fmt.Errorf("error retrieving group members: %w", err)	×
510	}	×
511	if len(members) == 0 {	12✔
512	return 0, nil // nothing to do	×
513	}	×
514
515	// prepare filter for upsert
516	currentTime := time.Now()	12✔
517		12✔
518	docs := make([]mongo.WriteModel, 0, len(members))	12✔
519	for _, member := range members {	58✔
520	// Create participant filter and document	46✔
521	censusParticipantsFilter := bson.M{	46✔
522	"participantID": member.ID,	46✔
523	"censusId": censusID,	46✔
524	}	46✔
525	participantDoc := &CensusParticipant{	46✔
526	ParticipantID: member.ID,	46✔
527	LoginHash: HashAuthTwoFaFields(*member, authFields, twoFaFields),	46✔
528	CensusID: censusID,	46✔
529	UpdatedAt: currentTime,	46✔
530	}	46✔
531		46✔
532	// Create participant update document	46✔
533	updateParticipantDoc, err := dynamicUpdateDocument(participantDoc, nil)	46✔
534	if err != nil {	46✔
535	log.Warnw("failed to create update document for participant",	×
NEW 536	"error", err, "participantID", member.ID)	×
537	continue	×
538	}
539
540	// Extract the $set part from the update document with type checking
541	setDoc, ok := updateParticipantDoc["$set"].(bson.M)	46✔
542	if !ok {	46✔
543	log.Warnw("failed to extract $set document for participant",	×
NEW 544	"error", "invalid $set type", "participantID", member.ID)	×
545	continue	×
546	}
547
548	// Create combined update document with both $set and $setOnInsert
549	combinedUpdateDoc := bson.M{	46✔
550	"$set": setDoc,	46✔
551	"$setOnInsert": bson.M{	46✔
552	"createdAt": currentTime,	46✔
553	},	46✔
554	}	46✔
555		46✔
556	// Create participant upsert model	46✔
557	upsertCensusParticipantsModel := mongo.NewUpdateOneModel().	46✔
558	SetFilter(censusParticipantsFilter).	46✔
559	SetUpdate(combinedUpdateDoc).	46✔
560	SetUpsert(true)	46✔
561	docs = append(docs, upsertCensusParticipantsModel)	46✔
562	}
563	// Unordered makes it continue on errors (e.g., one dup)
564	bulkOpts := options.BulkWrite().SetOrdered(false)	12✔
565		12✔
566	results, err := ms.censusParticipants.BulkWrite(ctx, docs, bulkOpts)	12✔
567	return results.UpsertedCount, err	12✔
568	}
569
570	// CensusParticipants retrieves all the census participants for a given census.
571	func (ms *MongoStorage) CensusParticipants(censusID internal.ObjectID) ([]CensusParticipant, error) {	7✔
572	// create a context with a timeout	7✔
573	ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)	7✔
574	defer cancel()	7✔
575		7✔
576	// validate input	7✔
577	if censusID.IsZero() {	7✔
578	return nil, ErrInvalidData	×
579	}	×
580
581	// prepare filter for upsert
582	filter := bson.M{	7✔
583	"censusId": censusID,	7✔
584	}	7✔
585		7✔
586	// find the participant	7✔
587	cursor, err := ms.censusParticipants.Find(ctx, filter)	7✔
588	if err != nil {	7✔
589	return nil, fmt.Errorf("failed to get census participants: %w", err)	×
590	}	×
591	defer func() {	14✔
592	if err := cursor.Close(ctx); err != nil {	7✔
593	log.Warnw("error closing cursor", "error", err)	×
594	}	×
595	}()
596	var participants []CensusParticipant	7✔
597	if err := cursor.All(ctx, &participants); err != nil {	7✔
598	return nil, fmt.Errorf("failed to get census participants: %w", err)	×
599	}	×
600
601	return participants, nil	7✔
602	}

vocdoni / saas-backend / 17557469823

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous