15452033010

Committed 04 Jun 2025 08:25PM UTC coverage: 55.908% (+1.3%) from 54.574%

Build # 15452033010

Build Type

Pull #115

github

Committed by

emmdim

Commit Message

feat: add organization member groups functionality

    - Introduced endpoints for managing organization member groups, including creation, retrieval, updating, and deletion.
    - Implemented MongoDB storage methods for organization member groups, including validation of member IDs.
    - Added types and structures for organization member groups in the database schema.
    - Created tests for organization member groups to ensure functionality and data integrity.

Pull Request Pull Request #115: Adds orgParticipant Groups Entity and API

Run Details

411 of 547 new or added lines in 7 files covered. (75.14%)

38 existing lines in 1 file now uncovered.

4736 of 8471 relevant lines covered (55.91%)

24.42 hits per line

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

82.53

/db/org_participants.go

package db

import (
        "context"
        "fmt"
        "math"
        "time"

        "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"
)

// CreateOrgParticipants creates a new orgParticipants for an organization
// reqires an existing organization
func (ms *MongoStorage) SetOrgParticipant(salt string, orgParticipant *OrgParticipant) (string, error) {
        // create a context with a timeout
        ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)
        defer cancel()

        if len(orgParticipant.OrgAddress) == 0 {
                return "", ErrInvalidData
        }

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

        if orgParticipant.Email != "" {
                // store only the hashed email
                orgParticipant.HashedEmail = internal.HashOrgData(orgParticipant.OrgAddress, orgParticipant.Email)
                orgParticipant.Email = ""
        }
        if orgParticipant.Phone != "" {
                // normalize and store only the hashed phone
                normalizedPhone, err := internal.SanitizeAndVerifyPhoneNumber(orgParticipant.Phone)
                if err == nil {
                        orgParticipant.HashedPhone = internal.HashOrgData(orgParticipant.OrgAddress, normalizedPhone)
                }
                orgParticipant.Phone = ""
        }
        if orgParticipant.Password != "" {
                // store only the hashed password
                orgParticipant.HashedPass = internal.HashPassword(salt, orgParticipant.Password)
                orgParticipant.Password = ""
        }

        if orgParticipant.ID != primitive.NilObjectID {
                // if the orgParticipant exists, update it with the new data
                orgParticipant.UpdatedAt = time.Now()
        } else {
                // if the orgParticipant doesn't exist, create the corresponding id
                orgParticipant.ID = primitive.NewObjectID()
                orgParticipant.CreatedAt = time.Now()
        }
        updateDoc, err := dynamicUpdateDocument(orgParticipant, nil)
        if err != nil {
                return "", err
        }
        ms.keysLock.Lock()
        defer ms.keysLock.Unlock()
        filter := bson.M{"_id": orgParticipant.ID}
        opts := options.Update().SetUpsert(true)
        _, err = ms.orgParticipants.UpdateOne(ctx, filter, updateDoc, opts)
        if err != nil {
                return "", err
        }

        return orgParticipant.ID.Hex(), nil
}

// DeleteOrgParticipants removes a orgParticipants and all its participants
func (ms *MongoStorage) DelOrgParticipant(id string) error {
        objID, err := primitive.ObjectIDFromHex(id)
        if err != nil {
                return ErrInvalidData
        }

        ms.keysLock.Lock()
        defer ms.keysLock.Unlock()
        // create a context with a timeout
        ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)
        defer cancel()

        // delete the orgParticipants from the database using the ID
        filter := bson.M{"_id": objID}
        _, err = ms.orgParticipants.DeleteOne(ctx, filter)
        return err
}

// OrgParticipant retrieves a orgParticipant from the DB based on it ID
func (ms *MongoStorage) OrgParticipant(id string) (*OrgParticipant, error) {
        objID, err := primitive.ObjectIDFromHex(id)
        if err != nil {
                return nil, ErrInvalidData
        }

        // create a context with a timeout
        ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)
        defer cancel()

        orgParticipant := &OrgParticipant{}
        if err = ms.orgParticipants.FindOne(ctx, bson.M{"_id": objID}).Decode(orgParticipant); err != nil {
                return nil, fmt.Errorf("failed to get orgParticipants: %w", err)
        }

        return orgParticipant, nil
}

// OrgParticipantByNo retrieves a orgParticipant from the DB based on organization address and participant number
func (ms *MongoStorage) OrgParticipantByNo(orgAddress, participantNo string) (*OrgParticipant, error) {
        if len(participantNo) == 0 {
                return nil, ErrInvalidData
        }
        // create a context with a timeout
        ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)
        defer cancel()

        orgParticipant := &OrgParticipant{}
        if err := ms.orgParticipants.FindOne(
                ctx, bson.M{"orgAddress": orgAddress, "participantNo": participantNo},
        ).Decode(orgParticipant); err != nil {
                return nil, fmt.Errorf("failed to get orgParticipants: %w", err)
        }

        return orgParticipant, nil
}

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

// validateBulkOrgParticipants validates the input parameters for bulk org participants
func (ms *MongoStorage) validateBulkOrgParticipants(
        orgAddress string,
        orgParticipants []OrgParticipant,
) (*Organization, error) {
        // Early returns for invalid input
        if len(orgParticipants) == 0 {
                return nil, nil // Not an error, just no work to do
        }
        if len(orgAddress) == 0 {
                return nil, ErrInvalidData
        }

        // Check that the organization exists
        org, err := ms.Organization(orgAddress)
        if err != nil {
                return nil, err
        }

        return org, nil
}

// prepareOrgParticipant processes a participant for storage
func prepareOrgParticipant(participant *OrgParticipant, orgAddress, salt string, currentTime time.Time) {
        participant.OrgAddress = orgAddress
        participant.CreatedAt = currentTime

        // Hash email if valid
        if participant.Email != "" {
                participant.HashedEmail = internal.HashOrgData(orgAddress, participant.Email)
                participant.Email = ""
        }

        // Hash phone if valid
        if participant.Phone != "" {
                normalizedPhone, err := internal.SanitizeAndVerifyPhoneNumber(participant.Phone)
                if err == nil {
                        participant.HashedPhone = internal.HashOrgData(orgAddress, normalizedPhone)
                }
                participant.Phone = ""
        }

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

// createOrgParticipantBulkOperations creates the bulk write operations for participants
func createOrgParticipantBulkOperations(
        participants []OrgParticipant,
        orgAddress string,
        salt string,
        currentTime time.Time,
) []mongo.WriteModel {
        var bulkOps []mongo.WriteModel

        for _, participant := range participants {
                // Prepare the participant
                prepareOrgParticipant(&participant, orgAddress, salt, currentTime)

                // Create filter and update document
                filter := bson.M{
                        "participantNo": participant.ParticipantNo,
                        "orgAddress":    orgAddress,
                }

                updateDoc, err := dynamicUpdateDocument(participant, nil)
                if err != nil {
                        log.Warnw("failed to create update document for participant",
                                "error", err, "participantNo", participant.ParticipantNo)
                        continue // Skip this participant but continue with others
                }

                // Create upsert model
                upsertModel := mongo.NewUpdateOneModel().
                        SetFilter(filter).
                        SetUpdate(updateDoc).
                        SetUpsert(true)
                bulkOps = append(bulkOps, upsertModel)
        }

        return bulkOps
}

// processOrgParticipantBatch processes a batch of participants and returns the number added
func (ms *MongoStorage) processOrgParticipantBatch(
        bulkOps []mongo.WriteModel,
) int {
        if len(bulkOps) == 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(), 20*time.Second)
        defer batchCancel()

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

        return len(bulkOps)
}

// startOrgParticipantProgressReporter starts a goroutine that reports progress periodically
func startOrgParticipantProgressReporter(
        ctx context.Context,
        progressChan chan<- *BulkOrgParticipantsStatus,
        totalParticipants int,
        processedParticipants *int,
        addedParticipants *int,
) {
        ticker := time.NewTicker(10 * time.Second)
        defer ticker.Stop()

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

// processOrgParticipantBatches processes participants in batches and sends progress updates
func (ms *MongoStorage) processOrgParticipantBatches(
        orgParticipants []OrgParticipant,
        orgAddress string,
        salt string,
        progressChan chan<- *BulkOrgParticipantsStatus,
) {
        defer close(progressChan)

        // Process participants in batches of 200
        batchSize := 200
        totalParticipants := len(orgParticipants)
        processedParticipants := 0
        addedParticipants := 0
        currentTime := time.Now()

        // Send initial progress
        progressChan <- &BulkOrgParticipantsStatus{
                Progress: 0,
                Total:    totalParticipants,
                Added:    addedParticipants,
        }

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

        // Start progress reporter in a separate goroutine
        go startOrgParticipantProgressReporter(
                ctx,
                progressChan,
                totalParticipants,
                &processedParticipants,
                &addedParticipants,
        )

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

                // Create bulk operations for this batch
                bulkOps := createOrgParticipantBulkOperations(
                        orgParticipants[i:end],
                        orgAddress,
                        salt,
                        currentTime,
                )

                // Process the batch and get number of added participants
                added := ms.processOrgParticipantBatch(bulkOps)
                addedParticipants += added

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

        // Send final progress (100%)
        progressChan <- &BulkOrgParticipantsStatus{
                Progress: 100,
                Total:    totalParticipants,
                Added:    addedParticipants,
        }
}

// SetBulkOrgParticipants adds multiple organization participants to the database in batches of 200 entries
// and updates already existing participants (decided by combination of participantNo and orgAddress)
// Requires an existing organization
// Returns a channel that sends the percentage of participants processed every 10 seconds.
// This function must be called in a goroutine.
func (ms *MongoStorage) SetBulkOrgParticipants(
        orgAddress, salt string,
        orgParticipants []OrgParticipant,
) (chan *BulkOrgParticipantsStatus, error) {
        progressChan := make(chan *BulkOrgParticipantsStatus, 10)

        // Validate input parameters
        org, err := ms.validateBulkOrgParticipants(orgAddress, orgParticipants)
        if err != nil {
                close(progressChan)
                return progressChan, err
        }

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

        // Start processing in a goroutine
        go ms.processOrgParticipantBatches(orgParticipants, orgAddress, salt, progressChan)

        return progressChan, nil
}

// OrgParticipantsWithPagination retrieves paginated orgParticipants for an organization from the DB
func (ms *MongoStorage) OrgParticipants(orgAddress string, page, pageSize int) ([]OrgParticipant, error) {
        if len(orgAddress) == 0 {
                return nil, ErrInvalidData
        }
        // create a context with a timeout
        ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)
        defer cancel()

        // Calculate skip value based on page and pageSize
        skip := (page - 1) * pageSize

        // Create filter
        filter := bson.M{"orgAddress": orgAddress}

        // Set up options for pagination
        findOptions := options.Find().
                SetSkip(int64(skip)).
                SetLimit(int64(pageSize))

        // Execute the find operation with pagination
        cursor, err := ms.orgParticipants.Find(ctx, filter, findOptions)
        if err != nil {
                return nil, fmt.Errorf("failed to get orgParticipants: %w", err)
        }
        defer func() {
                if err := cursor.Close(ctx); err != nil {
                        log.Warnw("error closing cursor", "error", err)
                }
        }()

        // Decode results
        var orgParticipants []OrgParticipant
        if err = cursor.All(ctx, &orgParticipants); err != nil {
                return nil, fmt.Errorf("failed to decode orgParticipants: %w", err)
        }

        return orgParticipants, nil
}

func (ms *MongoStorage) DeleteOrgParticipants(orgAddress string, participantIDs []string) (int, error) {
        if len(orgAddress) == 0 {
                return 0, ErrInvalidData
        }
        if len(participantIDs) == 0 {
                return 0, nil
        }

        // create a context with a timeout
        ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
        defer cancel()

        // create the filter for the delete operation
        filter := bson.M{
                "orgAddress": orgAddress,
                "participantNo": bson.M{
                        "$in": participantIDs,
                },
        }

        result, err := ms.orgParticipants.DeleteMany(ctx, filter)
        if err != nil {
                return 0, fmt.Errorf("failed to delete orgParticipants: %w", err)
        }

        return int(result.DeletedCount), nil
}

// validateOrgParticipants checks if the provided member IDs are valid
func (ms *MongoStorage) validateOrgMembers(ctx context.Context, orgAddress string, members []string) error {
        if len(members) == 0 {
                return fmt.Errorf("no members provided")
        }

        // Convert string IDs to ObjectIDs
        var objectIDs []primitive.ObjectID
        for _, id := range members {
                objID, err := primitive.ObjectIDFromHex(id)
                if err != nil {
                        return fmt.Errorf("invalid ObjectID format: %s", id)
                }
                objectIDs = append(objectIDs, objID)
        }

        cursor, err := ms.orgParticipants.Find(ctx, bson.M{
                "_id":        bson.M{"$in": objectIDs},
                "orgAddress": orgAddress,
        })
        if err != nil {
                return err
        }
        defer func() {
                if err := cursor.Close(ctx); err != nil {
                        log.Warnw("error closing cursor", "error", err)
                }
        }()

        var found []OrgParticipant
        if err := cursor.All(ctx, &found); err != nil {
                return err
        }

        // Create a map of found IDs for quick lookup
        foundMap := make(map[string]bool)
        for _, member := range found {
                foundMap[member.ID.Hex()] = true
        }

        // Check if all requested IDs were found
        for _, id := range members {
                if !foundMap[id] {
                        return fmt.Errorf("invalid member ID in add list: %s", id)
                }
        }
        return nil
}

// getOrgMembersByIDs retrieves organization members by their IDs
func (ms *MongoStorage) orgMembersByIDs(
        orgAddress string,
        memberIDs []string,
        page, pageSize int64,
) (int, []*OrgParticipant, error) {
        if len(memberIDs) == 0 {
                return 0, nil, nil // No members to retrieve
        }

        // create a context with a timeout
        ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)
        defer cancel()

        // Convert string IDs to ObjectIDs
        var objectIDs []primitive.ObjectID
        for _, id := range memberIDs {
                objID, err := primitive.ObjectIDFromHex(id)
                if err != nil {
                        return 0, nil, fmt.Errorf("invalid ObjectID format: %s", id)
                }
                objectIDs = append(objectIDs, objID)
        }

        filter := bson.M{
                "_id":        bson.M{"$in": objectIDs},
                "orgAddress": orgAddress,
        }

        // Count total documents
        totalCount, err := ms.orgParticipants.CountDocuments(ctx, filter)
        if err != nil {
                return 0, nil, err
        }
        totalPages := int(math.Ceil(float64(totalCount) / float64(pageSize)))

        // Calculate skip value based on page and pageSize
        skip := (page - 1) * pageSize

        // Set up options for pagination
        findOptions := options.Find().
                SetSkip(skip).
                SetLimit(pageSize)

        cursor, err := ms.orgParticipants.Find(ctx, filter, findOptions)
        if err != nil {
                return 0, nil, fmt.Errorf("failed to find org participants: %w", err)
        }
        defer func() {
                if err := cursor.Close(ctx); err != nil {
                        log.Warnw("error closing cursor", "error", err)
                }
        }()

        var members []*OrgParticipant
        if err := cursor.All(ctx, &members); err != nil {
                return 0, nil, fmt.Errorf("failed to decode org participants: %w", err)
        }

        return totalPages, members, nil
}

1	package db
2
3	import (
4	"context"
5	"fmt"
6	"math"
7	"time"
8
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	// CreateOrgParticipants creates a new orgParticipants for an organization
18	// reqires an existing organization
19	func (ms MongoStorage) SetOrgParticipant(salt string, orgParticipant OrgParticipant) (string, error) {	35✔
20	// create a context with a timeout	35✔
21	ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)	35✔
22	defer cancel()	35✔
23		35✔
24	if len(orgParticipant.OrgAddress) == 0 {	35✔
25	return "", ErrInvalidData	×
UNCOV 26	}	×
27
28	// check that the org exists
29	_, err := ms.Organization(orgParticipant.OrgAddress)	35✔
30	if err != nil {	35✔
31	if err == ErrNotFound {	×
32	return "", ErrInvalidData	×
33	}	×
UNCOV 34	return "", fmt.Errorf("organization not found: %w", err)	×
35	}
36
37	if orgParticipant.Email != "" {	68✔
38	// store only the hashed email	33✔
39	orgParticipant.HashedEmail = internal.HashOrgData(orgParticipant.OrgAddress, orgParticipant.Email)	33✔
40	orgParticipant.Email = ""	33✔
41	}	33✔
42	if orgParticipant.Phone != "" {	38✔
43	// normalize and store only the hashed phone	3✔
44	normalizedPhone, err := internal.SanitizeAndVerifyPhoneNumber(orgParticipant.Phone)	3✔
45	if err == nil {	6✔
46	orgParticipant.HashedPhone = internal.HashOrgData(orgParticipant.OrgAddress, normalizedPhone)	3✔
47	}	3✔
48	orgParticipant.Phone = ""	3✔
49	}
50	if orgParticipant.Password != "" {	37✔
51	// store only the hashed password	2✔
52	orgParticipant.HashedPass = internal.HashPassword(salt, orgParticipant.Password)	2✔
53	orgParticipant.Password = ""	2✔
54	}	2✔
55
56	if orgParticipant.ID != primitive.NilObjectID {	37✔
57	// if the orgParticipant exists, update it with the new data	2✔
58	orgParticipant.UpdatedAt = time.Now()	2✔
59	} else {	35✔
60	// if the orgParticipant doesn't exist, create the corresponding id	33✔
61	orgParticipant.ID = primitive.NewObjectID()	33✔
62	orgParticipant.CreatedAt = time.Now()	33✔
63	}	33✔
64	updateDoc, err := dynamicUpdateDocument(orgParticipant, nil)	35✔
65	if err != nil {	35✔
66	return "", err	×
UNCOV 67	}	×
68	ms.keysLock.Lock()	35✔
69	defer ms.keysLock.Unlock()	35✔
70	filter := bson.M{"_id": orgParticipant.ID}	35✔
71	opts := options.Update().SetUpsert(true)	35✔
72	_, err = ms.orgParticipants.UpdateOne(ctx, filter, updateDoc, opts)	35✔
73	if err != nil {	36✔
74	return "", err	1✔
75	}	1✔
76
77	return orgParticipant.ID.Hex(), nil	34✔
78	}
79
80	// DeleteOrgParticipants removes a orgParticipants and all its participants
81	func (ms *MongoStorage) DelOrgParticipant(id string) error {	2✔
82	objID, err := primitive.ObjectIDFromHex(id)	2✔
83	if err != nil {	3✔
84	return ErrInvalidData	1✔
85	}	1✔
86
87	ms.keysLock.Lock()	1✔
88	defer ms.keysLock.Unlock()	1✔
89	// create a context with a timeout	1✔
90	ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)	1✔
91	defer cancel()	1✔
92		1✔
93	// delete the orgParticipants from the database using the ID	1✔
94	filter := bson.M{"_id": objID}	1✔
95	_, err = ms.orgParticipants.DeleteOne(ctx, filter)	1✔
96	return err	1✔
97	}
98
99	// OrgParticipant retrieves a orgParticipant from the DB based on it ID
100	func (ms MongoStorage) OrgParticipant(id string) (OrgParticipant, error) {	7✔
101	objID, err := primitive.ObjectIDFromHex(id)	7✔
102	if err != nil {	8✔
103	return nil, ErrInvalidData	1✔
104	}	1✔
105
106	// create a context with a timeout
107	ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)	6✔
108	defer cancel()	6✔
109		6✔
110	orgParticipant := &OrgParticipant{}	6✔
111	if err = ms.orgParticipants.FindOne(ctx, bson.M{"_id": objID}).Decode(orgParticipant); err != nil {	8✔
112	return nil, fmt.Errorf("failed to get orgParticipants: %w", err)	2✔
113	}	2✔
114
115	return orgParticipant, nil	4✔
116	}
117
118	// OrgParticipantByNo retrieves a orgParticipant from the DB based on organization address and participant number
119	func (ms MongoStorage) OrgParticipantByNo(orgAddress, participantNo string) (OrgParticipant, error) {	19✔
120	if len(participantNo) == 0 {	19✔
121	return nil, ErrInvalidData	×
UNCOV 122	}	×
123	// create a context with a timeout
124	ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)	19✔
125	defer cancel()	19✔
126		19✔
127	orgParticipant := &OrgParticipant{}	19✔
128	if err := ms.orgParticipants.FindOne(	19✔
129	ctx, bson.M{"orgAddress": orgAddress, "participantNo": participantNo},	19✔
130	).Decode(orgParticipant); err != nil {	20✔
131	return nil, fmt.Errorf("failed to get orgParticipants: %w", err)	1✔
132	}	1✔
133
134	return orgParticipant, nil	18✔
135	}
136
137	// BulkOrgParticipantsStatus is returned by SetBulkOrgParticipants to provide the output.
138	type BulkOrgParticipantsStatus struct {
139	Progress int `json:"progress"`
140	Total int `json:"total"`
141	Added int `json:"added"`
142	}
143
144	// validateBulkOrgParticipants validates the input parameters for bulk org participants
145	func (ms *MongoStorage) validateBulkOrgParticipants(
146	orgAddress string,
147	orgParticipants []OrgParticipant,
148	) (*Organization, error) {	7✔
149	// Early returns for invalid input	7✔
150	if len(orgParticipants) == 0 {	7✔
151	return nil, nil // Not an error, just no work to do	×
UNCOV 152	}	×
153	if len(orgAddress) == 0 {	8✔
154	return nil, ErrInvalidData	1✔
155	}	1✔
156
157	// Check that the organization exists
158	org, err := ms.Organization(orgAddress)	6✔
159	if err != nil {	6✔
160	return nil, err	×
UNCOV 161	}	×
162
163	return org, nil	6✔
164	}
165
166	// prepareOrgParticipant processes a participant for storage
167	func prepareOrgParticipant(participant *OrgParticipant, orgAddress, salt string, currentTime time.Time) {	11✔
168	participant.OrgAddress = orgAddress	11✔
169	participant.CreatedAt = currentTime	11✔
170		11✔
171	// Hash email if valid	11✔
172	if participant.Email != "" {	22✔
173	participant.HashedEmail = internal.HashOrgData(orgAddress, participant.Email)	11✔
174	participant.Email = ""	11✔
175	}	11✔
176
177	// Hash phone if valid
178	if participant.Phone != "" {	22✔
179	normalizedPhone, err := internal.SanitizeAndVerifyPhoneNumber(participant.Phone)	11✔
180	if err == nil {	22✔
181	participant.HashedPhone = internal.HashOrgData(orgAddress, normalizedPhone)	11✔
182	}	11✔
183	participant.Phone = ""	11✔
184	}
185
186	// Hash password if present
187	if participant.Password != "" {	22✔
188	participant.HashedPass = internal.HashPassword(salt, participant.Password)	11✔
189	participant.Password = ""	11✔
190	}	11✔
191	}
192
193	// createOrgParticipantBulkOperations creates the bulk write operations for participants
194	func createOrgParticipantBulkOperations(
195	participants []OrgParticipant,
196	orgAddress string,
197	salt string,
198	currentTime time.Time,
199	) []mongo.WriteModel {	6✔
200	var bulkOps []mongo.WriteModel	6✔
201		6✔
202	for _, participant := range participants {	17✔
203	// Prepare the participant	11✔
204	prepareOrgParticipant(&participant, orgAddress, salt, currentTime)	11✔
205		11✔
206	// Create filter and update document	11✔
207	filter := bson.M{	11✔
208	"participantNo": participant.ParticipantNo,	11✔
209	"orgAddress": orgAddress,	11✔
210	}	11✔
211		11✔
212	updateDoc, err := dynamicUpdateDocument(participant, nil)	11✔
213	if err != nil {	11✔
214	log.Warnw("failed to create update document for participant",	×
215	"error", err, "participantNo", participant.ParticipantNo)	×
UNCOV 216	continue // Skip this participant but continue with others	×
217	}
218
219	// Create upsert model
220	upsertModel := mongo.NewUpdateOneModel().	11✔
221	SetFilter(filter).	11✔
222	SetUpdate(updateDoc).	11✔
223	SetUpsert(true)	11✔
224	bulkOps = append(bulkOps, upsertModel)	11✔
225	}
226
227	return bulkOps	6✔
228	}
229
230	// processOrgParticipantBatch processes a batch of participants and returns the number added
231	func (ms *MongoStorage) processOrgParticipantBatch(
232	bulkOps []mongo.WriteModel,
233	) int {	6✔
234	if len(bulkOps) == 0 {	6✔
235	return 0	×
UNCOV 236	}	×
237
238	// Only lock the mutex during the actual database operations
239	ms.keysLock.Lock()	6✔
240	defer ms.keysLock.Unlock()	6✔
241		6✔
242	// Create a new context for the batch	6✔
243	batchCtx, batchCancel := context.WithTimeout(context.Background(), 20*time.Second)	6✔
244	defer batchCancel()	6✔
245		6✔
246	// Execute the bulk write operations	6✔
247	_, err := ms.orgParticipants.BulkWrite(batchCtx, bulkOps)	6✔
248	if err != nil {	6✔
249	log.Warnw("failed to perform bulk operation on participants", "error", err)	×
250	return 0	×
UNCOV 251	}	×
252
253	return len(bulkOps)	6✔
254	}
255
256	// startOrgParticipantProgressReporter starts a goroutine that reports progress periodically
257	func startOrgParticipantProgressReporter(
258	ctx context.Context,
259	progressChan chan<- *BulkOrgParticipantsStatus,
260	totalParticipants int,
261	processedParticipants *int,
262	addedParticipants *int,
263	) {	6✔
264	ticker := time.NewTicker(10 * time.Second)	6✔
265	defer ticker.Stop()	6✔
266		6✔
267	for {	12✔
268	select {	6✔
269	case <-ticker.C:	×
270	// Calculate and send progress percentage	×
271	if totalParticipants > 0 {	×
272	progress := (processedParticipants 100) / totalParticipants	×
273	progressChan <- &BulkOrgParticipantsStatus{	×
274	Progress: progress,	×
275	Total: totalParticipants,	×
276	Added: *addedParticipants,	×
277	}	×
UNCOV 278	}	×
279	case <-ctx.Done():	6✔
280	return	6✔
281	}
282	}
283	}
284
285	// processOrgParticipantBatches processes participants in batches and sends progress updates
286	func (ms *MongoStorage) processOrgParticipantBatches(
287	orgParticipants []OrgParticipant,
288	orgAddress string,
289	salt string,
290	progressChan chan<- *BulkOrgParticipantsStatus,
291	) {	6✔
292	defer close(progressChan)	6✔
293		6✔
294	// Process participants in batches of 200	6✔
295	batchSize := 200	6✔
296	totalParticipants := len(orgParticipants)	6✔
297	processedParticipants := 0	6✔
298	addedParticipants := 0	6✔
299	currentTime := time.Now()	6✔
300		6✔
301	// Send initial progress	6✔
302	progressChan <- &BulkOrgParticipantsStatus{	6✔
303	Progress: 0,	6✔
304	Total: totalParticipants,	6✔
305	Added: addedParticipants,	6✔
306	}	6✔
307		6✔
308	// Create a context for the entire operation	6✔
309	ctx, cancel := context.WithCancel(context.Background())	6✔
310	defer cancel()	6✔
311		6✔
312	// Start progress reporter in a separate goroutine	6✔
313	go startOrgParticipantProgressReporter(	6✔
314	ctx,	6✔
315	progressChan,	6✔
316	totalParticipants,	6✔
317	&processedParticipants,	6✔
318	&addedParticipants,	6✔
319	)	6✔
320		6✔
321	// Process participants in batches	6✔
322	for i := 0; i < totalParticipants; i += batchSize {	12✔
323	// Calculate end index for current batch	6✔
324	end := i + batchSize	6✔
325	if end > totalParticipants {	12✔
326	end = totalParticipants	6✔
327	}	6✔
328
329	// Create bulk operations for this batch
330	bulkOps := createOrgParticipantBulkOperations(	6✔
331	orgParticipants[i:end],	6✔
332	orgAddress,	6✔
333	salt,	6✔
334	currentTime,	6✔
335	)	6✔
336		6✔
337	// Process the batch and get number of added participants	6✔
338	added := ms.processOrgParticipantBatch(bulkOps)	6✔
339	addedParticipants += added	6✔
340		6✔
341	// Update processed count	6✔
342	processedParticipants += (end - i)	6✔
343	}
344
345	// Send final progress (100%)
346	progressChan <- &BulkOrgParticipantsStatus{	6✔
347	Progress: 100,	6✔
348	Total: totalParticipants,	6✔
349	Added: addedParticipants,	6✔
350	}	6✔
351	}
352
353	// SetBulkOrgParticipants adds multiple organization participants to the database in batches of 200 entries
354	// and updates already existing participants (decided by combination of participantNo and orgAddress)
355	// Requires an existing organization
356	// Returns a channel that sends the percentage of participants processed every 10 seconds.
357	// This function must be called in a goroutine.
358	func (ms *MongoStorage) SetBulkOrgParticipants(
359	orgAddress, salt string,
360	orgParticipants []OrgParticipant,
361	) (chan *BulkOrgParticipantsStatus, error) {	7✔
362	progressChan := make(chan *BulkOrgParticipantsStatus, 10)	7✔
363		7✔
364	// Validate input parameters	7✔
365	org, err := ms.validateBulkOrgParticipants(orgAddress, orgParticipants)	7✔
366	if err != nil {	8✔
367	close(progressChan)	1✔
368	return progressChan, err	1✔
369	}	1✔
370
371	// If no participants, return empty channel
372	if org == nil {	6✔
373	close(progressChan)	×
374	return progressChan, nil	×
UNCOV 375	}	×
376
377	// Start processing in a goroutine
378	go ms.processOrgParticipantBatches(orgParticipants, orgAddress, salt, progressChan)	6✔
379		6✔
380	return progressChan, nil	6✔
381	}
382
383	// OrgParticipantsWithPagination retrieves paginated orgParticipants for an organization from the DB
384	func (ms *MongoStorage) OrgParticipants(orgAddress string, page, pageSize int) ([]OrgParticipant, error) {	7✔
385	if len(orgAddress) == 0 {	7✔
386	return nil, ErrInvalidData	×
UNCOV 387	}	×
388	// create a context with a timeout
389	ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)	7✔
390	defer cancel()	7✔
391		7✔
392	// Calculate skip value based on page and pageSize	7✔
393	skip := (page - 1) * pageSize	7✔
394		7✔
395	// Create filter	7✔
396	filter := bson.M{"orgAddress": orgAddress}	7✔
397		7✔
398	// Set up options for pagination	7✔
399	findOptions := options.Find().	7✔
400	SetSkip(int64(skip)).	7✔
401	SetLimit(int64(pageSize))	7✔
402		7✔
403	// Execute the find operation with pagination	7✔
404	cursor, err := ms.orgParticipants.Find(ctx, filter, findOptions)	7✔
405	if err != nil {	7✔
406	return nil, fmt.Errorf("failed to get orgParticipants: %w", err)	×
UNCOV 407	}	×
408	defer func() {	14✔
409	if err := cursor.Close(ctx); err != nil {	7✔
410	log.Warnw("error closing cursor", "error", err)	×
UNCOV 411	}	×
412	}()
413
414	// Decode results
415	var orgParticipants []OrgParticipant	7✔
416	if err = cursor.All(ctx, &orgParticipants); err != nil {	7✔
417	return nil, fmt.Errorf("failed to decode orgParticipants: %w", err)	×
UNCOV 418	}	×
419
420	return orgParticipants, nil	7✔
421	}
422
423	func (ms *MongoStorage) DeleteOrgParticipants(orgAddress string, participantIDs []string) (int, error) {	2✔
424	if len(orgAddress) == 0 {	2✔
425	return 0, ErrInvalidData	×
UNCOV 426	}	×
427	if len(participantIDs) == 0 {	2✔
428	return 0, nil	×
UNCOV 429	}	×
430
431	// create a context with a timeout
432	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)	2✔
433	defer cancel()	2✔
434		2✔
435	// create the filter for the delete operation	2✔
436	filter := bson.M{	2✔
437	"orgAddress": orgAddress,	2✔
438	"participantNo": bson.M{	2✔
439	"$in": participantIDs,	2✔
440	},	2✔
441	}	2✔
442		2✔
443	result, err := ms.orgParticipants.DeleteMany(ctx, filter)	2✔
444	if err != nil {	2✔
445	return 0, fmt.Errorf("failed to delete orgParticipants: %w", err)	×
UNCOV 446	}	×
447
448	return int(result.DeletedCount), nil	2✔
449	}
450
451	// validateOrgParticipants checks if the provided member IDs are valid
452	func (ms *MongoStorage) validateOrgMembers(ctx context.Context, orgAddress string, members []string) error {	28✔
453	if len(members) == 0 {	28✔
NEW UNCOV 454	return fmt.Errorf("no members provided")	×
NEW UNCOV 455	}	×
456
457	// Convert string IDs to ObjectIDs
458	var objectIDs []primitive.ObjectID	28✔
459	for _, id := range members {	91✔
460	objID, err := primitive.ObjectIDFromHex(id)	63✔
461	if err != nil {	66✔
462	return fmt.Errorf("invalid ObjectID format: %s", id)	3✔
463	}	3✔
464	objectIDs = append(objectIDs, objID)	60✔
465	}
466
467	cursor, err := ms.orgParticipants.Find(ctx, bson.M{	25✔
468	"_id": bson.M{"$in": objectIDs},	25✔
469	"orgAddress": orgAddress,	25✔
470	})	25✔
471	if err != nil {	25✔
NEW UNCOV 472	return err	×
NEW UNCOV 473	}	×
474	defer func() {	50✔
475	if err := cursor.Close(ctx); err != nil {	25✔
NEW UNCOV 476	log.Warnw("error closing cursor", "error", err)	×
NEW UNCOV 477	}	×
478	}()
479
480	var found []OrgParticipant	25✔
481	if err := cursor.All(ctx, &found); err != nil {	25✔
NEW UNCOV 482	return err	×
NEW UNCOV 483	}	×
484
485	// Create a map of found IDs for quick lookup
486	foundMap := make(map[string]bool)	25✔
487	for _, member := range found {	85✔
488	foundMap[member.ID.Hex()] = true	60✔
489	}	60✔
490
491	// Check if all requested IDs were found
492	for _, id := range members {	85✔
493	if !foundMap[id] {	60✔
NEW UNCOV 494	return fmt.Errorf("invalid member ID in add list: %s", id)	×
NEW UNCOV 495	}	×
496	}
497	return nil	25✔
498	}
499
500	// getOrgMembersByIDs retrieves organization members by their IDs
501	func (ms *MongoStorage) orgMembersByIDs(
502	orgAddress string,
503	memberIDs []string,
504	page, pageSize int64,
505	) (int, []*OrgParticipant, error) {	6✔
506	if len(memberIDs) == 0 {	7✔
507	return 0, nil, nil // No members to retrieve	1✔
508	}	1✔
509
510	// create a context with a timeout
511	ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)	5✔
512	defer cancel()	5✔
513		5✔
514	// Convert string IDs to ObjectIDs	5✔
515	var objectIDs []primitive.ObjectID	5✔
516	for _, id := range memberIDs {	18✔
517	objID, err := primitive.ObjectIDFromHex(id)	13✔
518	if err != nil {	13✔
NEW UNCOV 519	return 0, nil, fmt.Errorf("invalid ObjectID format: %s", id)	×
NEW UNCOV 520	}	×
521	objectIDs = append(objectIDs, objID)	13✔
522	}
523
524	filter := bson.M{	5✔
525	"_id": bson.M{"$in": objectIDs},	5✔
526	"orgAddress": orgAddress,	5✔
527	}	5✔
528		5✔
529	// Count total documents	5✔
530	totalCount, err := ms.orgParticipants.CountDocuments(ctx, filter)	5✔
531	if err != nil {	5✔
NEW UNCOV 532	return 0, nil, err	×
NEW UNCOV 533	}	×
534	totalPages := int(math.Ceil(float64(totalCount) / float64(pageSize)))	5✔
535		5✔
536	// Calculate skip value based on page and pageSize	5✔
537	skip := (page - 1) * pageSize	5✔
538		5✔
539	// Set up options for pagination	5✔
540	findOptions := options.Find().	5✔
541	SetSkip(skip).	5✔
542	SetLimit(pageSize)	5✔
543		5✔
544	cursor, err := ms.orgParticipants.Find(ctx, filter, findOptions)	5✔
545	if err != nil {	5✔
NEW UNCOV 546	return 0, nil, fmt.Errorf("failed to find org participants: %w", err)	×
NEW UNCOV 547	}	×
548	defer func() {	10✔
549	if err := cursor.Close(ctx); err != nil {	5✔
NEW UNCOV 550	log.Warnw("error closing cursor", "error", err)	×
NEW UNCOV 551	}	×
552	}()
553
554	var members []*OrgParticipant	5✔
555	if err := cursor.All(ctx, &members); err != nil {	5✔
NEW UNCOV 556	return 0, nil, fmt.Errorf("failed to decode org participants: %w", err)	×
NEW UNCOV 557	}	×
558
559	return totalPages, members, nil	5✔
560	}

vocdoni / saas-backend / 15452033010

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