add publish doc events in bulkUpdate func

Author: rostikts

database/mongo/base.go

@@ -377,6 +377,26 @@ func (mg *Mongo) UpdateDocuments(auth model.Auth, dbName, col string, filters ma
 	secureWrite(acctID, userID, auth.Role, col, filters)
 	removeNotEditableFields(updateFields)
 
+	var ids []string
+	findOpts := options.Find().SetProjection(bson.D{{"id", 1}})
+	cur, err := db.Collection(internal.CleanCollectionName(col)).Find(mg.Ctx, filters, findOpts)
+	if err != nil {
+		return 0, err
+	}
+	for cur.Next(mg.Ctx) {
+		var v map[string]interface{}
+		if err := cur.Decode(&v); err != nil {
+			mg.log.Error().Err(err).Msg("")
+		}
+		id, ok := v[FieldID].(primitive.ObjectID)
+		if ok {
+			ids = append(ids, id.Hex())
+		}
+	}
+	if len(ids) == 0 {
+		return 0, nil
+	}
+
 	newProps := bson.M{}
 	for k, v := range updateFields {
 		newProps[k] = v
@@ -388,6 +408,17 @@ func (mg *Mongo) UpdateDocuments(auth model.Auth, dbName, col string, filters ma
 	if err != nil {
 		return 0, err
 	}
+
+	go func() {
+		for _, id := range ids {
+			doc, err := mg.GetDocumentByID(auth, dbName, col, id)
+			if err != nil {
+				mg.log.Error().Err(err).Msgf("the document with id=%s is not received for publishDocument event", id)
+				continue
+			}
+			mg.PublishDocument("db-"+col, internal.MsgTypeDBUpdated, doc)
+		}
+	}()
 	return res.ModifiedCount, err
 }
 

database/mongo/mongo_test.go

@@ -3,6 +3,7 @@ package mongo
 import (
 	"context"
 	"errors"
+	"github.com/staticbackendhq/core/logger"
 	"log"
 	"os"
 	"testing"
@@ -51,6 +52,7 @@ func TestMain(m *testing.M) {
 		Client:          cl,
 		Ctx:             context.Background(),
 		PublishDocument: fakePubDocEvent,
+		log:             logger.Get(config.Current),
 	}
 
 	if err := datastore.Ping(); err != nil {

database/postgresql/base.go

@@ -265,7 +265,30 @@ func (pg *PostgreSQL) UpdateDocuments(auth model.Auth, dbName, col string, filte
 	where := secureWrite(auth, col)
 	where = applyFilter(where, filters)
 
+	var idsForUpdate []string
 	qry := fmt.Sprintf(`
+		SELECT id
+		FROM %s.%s 
+		%s
+	`, dbName, internal.CleanCollectionName(col), where)
+
+	rows, err := pg.DB.Query(qry, auth.AccountID, auth.UserID)
+	if err != nil {
+		return
+	}
+	for rows.Next() {
+		var id string
+		if err := rows.Scan(&id); err != nil {
+			pg.log.Error().Err(err).Msg("error occurred during scanning id for UpdateDocument event")
+			continue
+		}
+		idsForUpdate = append(idsForUpdate, id)
+	}
+	if len(idsForUpdate) == 0 {
+		return 0, nil
+	}
+
+	qry = fmt.Sprintf(`
 		UPDATE %s.%s SET
 			data = data || $3
 		%s
@@ -283,6 +306,17 @@ func (pg *PostgreSQL) UpdateDocuments(auth model.Auth, dbName, col string, filte
 	if err != nil {
 		return 0, err
 	}
+
+	go func() {
+		for _, id := range idsForUpdate {
+			doc, err := pg.GetDocumentByID(auth, dbName, col, id)
+			if err != nil {
+				pg.log.Error().Err(err).Msgf("the document with id=%s is not received for publishDocument event", id)
+				continue
+			}
+			pg.PublishDocument("db-"+col, internal.MsgTypeDBUpdated, doc)
+		}
+	}()
 	return
 }