17262379833

Committed 27 Aug 2025 09:07AM UTC coverage: 58.431% (+0.4%) from 57.998%

Build # 17262379833

Build Type

Pull #206

github

Committed by

web-flow

Commit Message

f/csp refactor fixes (#217)

* fix AuthOnly flow
* adapt to changes introduced by new type Phone
* fix test race condition
* drop unused methods
* lint

Pull Request Pull Request #206: csp: Refactor csp to allow login with arbitrary authFields and twoFaFields

Run Details

132 of 180 new or added lines in 6 files covered. (73.33%)

21 existing lines in 4 files now uncovered.

5482 of 9382 relevant lines covered (58.43%)

28.14 hits per line

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

85.32

/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/bson/primitive"
        "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 len(participant.ParticipantID) == 0 || len(participant.CensusID) == 0 {
                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 string) (*CensusParticipant, error) {
        // create a context with a timeout
        ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)
        defer cancel()

        // validate input
        if len(id) == 0 || len(censusID) == 0 {
                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 string,
        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 || len(censusID) == 0 {
                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.Hex(),
                "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 string,
        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 || len(censusID) == 0 {
                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 string) 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 len(participantID) == 0 || len(censusID) == 0 {
                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"`
}

// prepareMember processes a member for storage by:
// - Setting the organization address
// - Setting the creation timestamp
// - Hashing sensitive data (email, phone, password)
// - Clearing the original sensitive data
func prepareMember(member *OrgMember, orgAddress common.Address, salt string, currentTime time.Time) {
        // Assign a new internal ID if not provided
        if member.ID == primitive.NilObjectID {
                member.ID = primitive.NewObjectID()
        }

        member.OrgAddress = orgAddress
        member.CreatedAt = currentTime

        // Phone handling is now done by the Phone type itself
        if member.Phone != nil && !member.Phone.IsEmpty() {
                // Ensure the phone has the correct org address for hashing
                member.Phone.HashWithOrgAddress(orgAddress)
        }

        // Hash password if present
        if member.Password != "" {
                member.HashedPass = internal.HashPassword(salt, member.Password)
                member.Password = ""
        }
}

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

        for _, orgMember := range orgMembers {
                // Prepare the member
                prepareMember(&orgMember, orgAddress, salt, currentTime)

                // Create member filter and update document
                memberFilter := bson.M{
                        "_id":        orgMember.ID,
                        "orgAddress": 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.Hex(),
                        "censusId":      censusID,
                }
                participantDoc := &CensusParticipant{
                        ParticipantID: orgMember.ID.Hex(),
                        CensusID:      censusID,
                        CreatedAt:     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.Hex())
                        continue
                }

                // Create participant upsert model
                upsertCensusParticipantsModel := mongo.NewUpdateOneModel().
                        SetFilter(censusParticipantsFilter).
                        SetUpdate(updateParticipantDoc).
                        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 string,
        orgMembersSize int,
) (*Census, error) {
        // Early returns for invalid input
        if orgMembersSize == 0 {
                return nil, nil // Not an error, just no work to do
        }
        if len(censusID) == 0 {
                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,
        census *Census,
        censusID string,
        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],
                        census.OrgAddress,
                        censusID,
                        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(
        salt, censusID string, 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, census, censusID, salt, progressChan)

        return progressChan, nil
}

func (ms *MongoStorage) setBulkCensusParticipant(
        ctx context.Context, censusID, groupID string, 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)
        }

        currentTime := time.Now()

        docs := make([]mongo.WriteModel, 0, len(members))
        for _, member := range members {
                // Create participant filter and document
                id := member.ID.Hex()
                censusParticipantsFilter := bson.M{
                        "participantID": id,
                        "censusId":      censusID,
                }
                participantDoc := &CensusParticipant{
                        ParticipantID: id,
                        LoginHash:     HashAuthTwoFaFields(*member, authFields, twoFaFields),
                        CensusID:      censusID,
                        CreatedAt:     currentTime,
                        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", id)
                        continue
                }

                // Create participant upsert model
                upsertCensusParticipantsModel := mongo.NewUpdateOneModel().
                        SetFilter(censusParticipantsFilter).
                        SetUpdate(updateParticipantDoc).
                        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 string) ([]CensusParticipant, error) {
        // create a context with a timeout
        ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)
        defer cancel()

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

vocdoni / saas-backend / 17262379833

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