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

83.83

/db/org_members.go

package db

import (
        "context"
        "fmt"
        "math"
        "net/mail"
        "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"
)

// SetOrgMember creates a new orgMembers for an organization
// requires an existing organization
func (ms *MongoStorage) SetOrgMember(salt string, orgMember *OrgMember) (string, error) {
        // create a context with a timeout
        ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)
        defer cancel()

        if orgMember.OrgAddress.Cmp(common.Address{}) == 0 {
                return "", ErrInvalidData
        }

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

        // Phone handling is now done by the Phone type itself
        if orgMember.Phone != nil && !orgMember.Phone.IsEmpty() {
                // Ensure the phone has the correct org address for hashing
                orgMember.Phone.HashWithOrgAddress(orgMember.OrgAddress)
        }
        if orgMember.Password != "" {
                // store only the hashed password
                orgMember.HashedPass = internal.HashPassword(salt, orgMember.Password)
                orgMember.Password = ""
        }

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

        return orgMember.ID.Hex(), nil
}

// DeleteOrgMember removes a orgMember
func (ms *MongoStorage) DelOrgMember(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 orgMember from the database using the ID
        filter := bson.M{"_id": objID}
        _, err = ms.orgMembers.DeleteOne(ctx, filter)
        return err
}

// OrgMember retrieves a orgMember from the DB based on it ID
func (ms *MongoStorage) OrgMember(orgAddress common.Address, id string) (*OrgMember, 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()

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

        return orgMember, nil
}

// OrgMemberByMemberNumber retrieves a orgMember from the DB based on organization address and member number
func (ms *MongoStorage) OrgMemberByMemberNumber(orgAddress common.Address, memberNumber string) (*OrgMember, error) {
        if len(memberNumber) == 0 {
                return nil, ErrInvalidData
        }
        // create a context with a timeout
        ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)
        defer cancel()

        orgMember := &OrgMember{}
        if err := ms.orgMembers.FindOne(
                ctx, bson.M{"orgAddress": orgAddress, "memberNumber": memberNumber},
        ).Decode(orgMember); err != nil {
                if err == mongo.ErrNoDocuments {
                        return nil, ErrNotFound
                }
                return nil, fmt.Errorf("failed to get orgMember: %w", err)
        }

        return orgMember, nil
}

// BulkOrgMembersJob is returned by SetBulkOrgMembers to provide the output.
type BulkOrgMembersJob struct {
        Progress int
        Total    int
        Added    int
        Errors   []error
}

func (j *BulkOrgMembersJob) ErrorsAsStrings() []string {
        s := []string{}
        for _, err := range j.Errors {
                s = append(s, err.Error())
        }
        return s
}

// validateBulkOrgMembers validates the input parameters for bulk org members
func (ms *MongoStorage) validateBulkOrgMembers(
        orgAddress common.Address,
        orgMembers []OrgMember,
) (*Organization, error) {
        // Early returns for invalid input
        if len(orgMembers) == 0 {
                return nil, nil // Not an error, just no work to do
        }
        if orgAddress.Cmp(common.Address{}) == 0 {
                return nil, ErrInvalidData
        }

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

        return org, nil
}

// prepareOrgMember processes a member for storage
func prepareOrgMember(member *OrgMember, orgAddress common.Address, salt string, currentTime time.Time) []error {
        var errors []error

        // Assign a new internal ID if not provided
        if member.ID == primitive.NilObjectID {
                member.ID = primitive.NewObjectID()
        }
        member.OrgAddress = orgAddress
        member.CreatedAt = currentTime

        // check if mail is valid
        if member.Email != "" {
                if _, err := mail.ParseAddress(member.Email); err != nil {
                        errors = append(errors, fmt.Errorf("could not parse from email: %s %v", member.Email, err))
                        // If email is invalid, set it to empty and store the error
                        member.Email = ""
                }
        }

        // Phone handling is now done by the Phone type itself
        if member.Phone != nil && !member.Phone.IsEmpty() {
                if err := member.Phone.Validate(); err != nil {
                        errors = append(errors, fmt.Errorf("invalid phone: %s %v", member.Phone, err))
                        member.Phone = nil
                } else {
                        // 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 = ""
        }

        // Check that the birthdate is valid
        if len(member.BirthDate) > 0 {
                if _, err := time.Parse("2006-01-02", member.BirthDate); err != nil {
                        errors = append(errors, fmt.Errorf("invalid birthdate format: %s %v", member.BirthDate, err))
                        member.BirthDate = "" // Reset invalid birthdate
                }
        }
        return errors
}

// createOrgMemberBulkOperations creates a batch of members using bulk write operations,
// and returns the number of members added (or updated) and any errors encountered.
func (ms *MongoStorage) createOrgMemberBulkOperations(
        members []OrgMember,
        orgAddress common.Address,
        salt string,
        currentTime time.Time,
) (int, []error) {
        var bulkOps []mongo.WriteModel
        var errors []error

        for _, member := range members {
                // Prepare the member
                validationErrors := prepareOrgMember(&member, orgAddress, salt, currentTime)
                errors = append(errors, validationErrors...)

                // Create filter for existing members and update document
                filter := bson.M{
                        "_id":        member.ID,
                        "orgAddress": orgAddress,
                }

                updateDoc, err := dynamicUpdateDocument(member, nil)
                if err != nil {
                        log.Warnw("failed to create update document for member",
                                "error", err, "ID", member.ID)
                        errors = append(errors, fmt.Errorf("member %s: %w", member.ID.Hex(), err))
                        continue // Skip this member but continue with others
                }

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

        if len(bulkOps) == 0 {
                return 0, errors
        }

        // 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
        result, err := ms.orgMembers.BulkWrite(batchCtx, bulkOps)
        if err != nil {
                log.Warnw("error during bulk operation on members batch", "error", err)
                firstID := members[0].ID
                lastID := members[len(members)-1].ID
                errors = append(errors, fmt.Errorf("batch %s - %s: %w", firstID.Hex(), lastID.Hex(), err))
        }

        return int(result.ModifiedCount + result.UpsertedCount), errors
}

// startOrgMemberProgressReporter starts a goroutine that reports progress periodically
func startOrgMemberProgressReporter(
        ctx context.Context,
        progressChan chan<- *BulkOrgMembersJob,
        status *BulkOrgMembersJob,
) {
        defer close(progressChan)

        if status.Total == 0 {
                return
        }

        ticker := time.NewTicker(10 * time.Second)
        defer ticker.Stop()

        // Send initial progress
        progressChan <- status

        for {
                select {
                case <-ticker.C:
                        progressChan <- status
                case <-ctx.Done():
                        // Send final progress (100%)
                        progressChan <- status
                        return
                }
        }
}

// processOrgMemberBatches processes members in batches and sends progress updates
func (ms *MongoStorage) processOrgMemberBatches(
        orgMembers []OrgMember,
        orgAddress common.Address,
        salt string,
        progressChan chan<- *BulkOrgMembersJob,
) {
        if len(orgMembers) == 0 {
                close(progressChan)
                return
        }

        // Process members in batches of 200
        batchSize := 200
        currentTime := time.Now()

        job := BulkOrgMembersJob{
                Progress: 0,
                Total:    len(orgMembers),
                Added:    0,
                Errors:   []error{},
        }

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

        // Start progress reporter in a separate goroutine
        go startOrgMemberProgressReporter(
                ctx,
                progressChan,
                &job,
        )

        // Process members in batches
        for start := 0; start < job.Total; start += batchSize {
                // Calculate end index for current batch
                end := min(start+batchSize, job.Total)

                // Process the batch and get number of added members
                added, errs := ms.createOrgMemberBulkOperations(
                        orgMembers[start:end],
                        orgAddress,
                        salt,
                        currentTime,
                )

                // Update job stats
                job = BulkOrgMembersJob{
                        Progress: int(float64(job.Added+added) / float64(job.Total) * 100),
                        Total:    job.Total,
                        Added:    job.Added + added,
                        Errors:   append(job.Errors, errs...),
                }
        }
}

// SetBulkOrgMembers adds multiple organization members to the database in batches of 200 entries
// and updates already existing members (decided by combination of internal id and orgAddress)
// Requires an existing organization
// Returns a channel that sends the percentage of members processed every 10 seconds.
// This function must be called in a goroutine.
func (ms *MongoStorage) SetBulkOrgMembers(
        orgAddress common.Address, salt string,
        orgMembers []OrgMember,
) (chan *BulkOrgMembersJob, error) {
        progressChan := make(chan *BulkOrgMembersJob, 10)

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

        // Start processing in a goroutine
        go ms.processOrgMemberBatches(orgMembers, orgAddress, salt, progressChan)

        return progressChan, nil
}

// OrgMembers retrieves paginated orgMembers for an organization from the DB
func (ms *MongoStorage) OrgMembers(orgAddress common.Address, page, pageSize int, search string) (int, []OrgMember, error) {
        if orgAddress.Cmp(common.Address{}) == 0 {
                return 0, nil, ErrInvalidData
        }
        // create a context with a timeout
        ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)
        defer cancel()

        // Create filter
        filter := bson.M{
                "orgAddress": orgAddress,
        }
        if len(search) > 0 {
                filter["$or"] = []bson.M{
                        {"email": bson.M{"$regex": search, "$options": "i"}},
                        {"memberNumber": bson.M{"$regex": search, "$options": "i"}},
                        {"nationalID": bson.M{"$regex": search, "$options": "i"}},
                        {"name": bson.M{"$regex": search, "$options": "i"}},
                        {"surname": bson.M{"$regex": search, "$options": "i"}},
                        {"birthDate": bson.M{"$regex": search, "$options": "i"}},
                }
        }

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

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

        sort := bson.D{
                bson.E{Key: "name", Value: 1},
                bson.E{Key: "surname", Value: 1},
        }
        // Set up options for pagination
        findOptions := options.Find().
                SetSort(sort). // Sort by createdAt in descending order
                SetSkip(int64(skip)).
                SetLimit(int64(pageSize))

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

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

        return totalPages, orgMembers, nil
}

func (ms *MongoStorage) DeleteOrgMembers(orgAddress common.Address, ids []string) (int, error) {
        if orgAddress.Cmp(common.Address{}) == 0 {
                return 0, ErrInvalidData
        }
        if len(ids) == 0 {
                return 0, nil
        }
        // Convert string IDs to ObjectIDs
        var oids []primitive.ObjectID
        for _, id := range ids {
                objID, err := primitive.ObjectIDFromHex(id)
                if err != nil {
                        return 0, fmt.Errorf("invalid member ID %s: %w", id, ErrInvalidData)
                }
                oids = append(oids, objID)
        }

        // 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,
                "_id": bson.M{
                        "$in": oids,
                },
        }

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

        return int(result.DeletedCount), nil
}

// validateOrgMembers checks if the provided member IDs are valid
func (ms *MongoStorage) validateOrgMembers(ctx context.Context, orgAddress common.Address, 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.orgMembers.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 []OrgMember
        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 common.Address,
        memberIDs []string,
        page, pageSize int64,
) (int, []*OrgMember, 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.orgMembers.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.orgMembers.Find(ctx, filter, findOptions)
        if err != nil {
                return 0, nil, fmt.Errorf("failed to find org members: %w", err)
        }
        defer func() {
                if err := cursor.Close(ctx); err != nil {
                        log.Warnw("error closing cursor", "error", err)
                }
        }()

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

        return totalPages, members, nil
}

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

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