diff --git a/compose/service/attachment.go b/compose/service/attachment.go index 0618286bf..b186f46f7 100644 --- a/compose/service/attachment.go +++ b/compose/service/attachment.go @@ -61,6 +61,7 @@ func Attachment(store files.Store) AttachmentService { return (&attachment{ files: store, ac: DefaultAccessControl, + store: DefaultNgStore, }).With(context.Background()) } @@ -70,6 +71,7 @@ func (svc attachment) With(ctx context.Context) AttachmentService { actionlog: DefaultActionlog, ac: svc.ac, files: svc.files, + store: svc.store, } } diff --git a/messaging/repository/attachment.go b/messaging/repository/attachment.go index 6bce36d29..0ec0295cc 100644 --- a/messaging/repository/attachment.go +++ b/messaging/repository/attachment.go @@ -1,135 +1,135 @@ package repository -import ( - "context" - "time" - - "github.com/Masterminds/squirrel" - "github.com/titpetric/factory" - - "github.com/cortezaproject/corteza-server/messaging/types" - "github.com/cortezaproject/corteza-server/pkg/rh" -) - -type ( - AttachmentRepository interface { - With(ctx context.Context, db *factory.DB) AttachmentRepository - - FindAttachmentByID(id uint64) (*types.Attachment, error) - FindAttachmentByMessageID(IDs ...uint64) (types.AttachmentSet, error) - - CreateAttachment(mod *types.Attachment) (*types.Attachment, error) - DeleteAttachmentByID(id uint64) error - - BindAttachment(attachmentId, messageId uint64) error - } - - attachment struct { - *repository - } -) - -const ( - ErrAttachmentNotFound = repositoryError("AttachmentNotFound") -) - -func Attachment(ctx context.Context, db *factory.DB) AttachmentRepository { - return (&attachment{}).With(ctx, db) -} - -func (r attachment) With(ctx context.Context, db *factory.DB) AttachmentRepository { - return &attachment{ - repository: r.repository.With(ctx, db), - } -} - -func (r attachment) table() string { - return "messaging_attachment" -} - -func (r attachment) tableMessage() string { - return "messaging_message_attachment" -} - -func (r attachment) columns() []string { - return []string{ - "a.id", - "a.rel_user", - "a.url", - "a.preview_url", - "a.name", - "a.meta", - "a.created_at", - "a.updated_at", - "a.deleted_at", - } -} - -func (r attachment) query() squirrel.SelectBuilder { - return squirrel. - Select(r.columns()...). - From(r.table() + " AS a"). - Where("a.deleted_at IS NULL") - -} - -func (r attachment) FindAttachmentByID(ID uint64) (*types.Attachment, error) { - return r.findOneBy("id", ID) -} - -func (r attachment) findOneBy(field string, value interface{}) (*types.Attachment, error) { - var ( - p = &types.Attachment{} - - q = r.query(). - Where(squirrel.Eq{field: value}) - - err = rh.FetchOne(r.db(), q, p) - ) - - if err != nil { - return nil, err - } else if p.ID == 0 { - return nil, ErrAttachmentNotFound - } - - return p, nil -} - -func (r attachment) FindAttachmentByMessageID(IDs ...uint64) (rval types.AttachmentSet, err error) { - rval = types.AttachmentSet{} - - if len(IDs) == 0 { - return - } - - query := r.query(). - Columns("ma.rel_message"). - Join(r.tableMessage() + " AS ma ON (a.id = ma.rel_attachment)"). - Where(squirrel.Eq{"rel_message": IDs}) - - return rval, rh.FetchAll(r.db(), query, &rval) -} - -func (r attachment) CreateAttachment(mod *types.Attachment) (*types.Attachment, error) { - if mod.ID == 0 { - mod.ID = factory.Sonyflake.NextID() - } - - mod.CreatedAt = time.Now() - - return mod, r.db().Insert(r.table(), mod) -} - -func (r attachment) DeleteAttachmentByID(ID uint64) error { - return rh.UpdateColumns(r.db(), r.table(), rh.Set{"deleted_at": time.Now()}, squirrel.Eq{"id": ID}) -} - -func (r attachment) BindAttachment(attachmentId, messageId uint64) error { - bond := struct { - RelAttachment uint64 `db:"rel_attachment"` - RelMessage uint64 `db:"rel_message"` - }{attachmentId, messageId} - - return r.db().Insert(r.tableMessage(), bond) -} +//import ( +// "context" +// "time" +// +// "github.com/Masterminds/squirrel" +// "github.com/titpetric/factory" +// +// "github.com/cortezaproject/corteza-server/messaging/types" +// "github.com/cortezaproject/corteza-server/pkg/rh" +//) +// +//type ( +// AttachmentRepository interface { +// With(ctx context.Context, db *factory.DB) AttachmentRepository +// +// FindAttachmentByID(id uint64) (*types.Attachment, error) +// FindAttachmentByMessageID(IDs ...uint64) (types.AttachmentSet, error) +// +// CreateAttachment(mod *types.Attachment) (*types.Attachment, error) +// DeleteAttachmentByID(id uint64) error +// +// BindAttachment(attachmentId, messageId uint64) error +// } +// +// attachment struct { +// *repository +// } +//) +// +//const ( +// ErrAttachmentNotFound = repositoryError("AttachmentNotFound") +//) +// +//func Attachment(ctx context.Context, db *factory.DB) AttachmentRepository { +// return (&attachment{}).With(ctx, db) +//} +// +//func (r attachment) With(ctx context.Context, db *factory.DB) AttachmentRepository { +// return &attachment{ +// repository: r.repository.With(ctx, db), +// } +//} +// +//func (r attachment) table() string { +// return "messaging_attachment" +//} +// +//func (r attachment) tableMessage() string { +// return "messaging_message_attachment" +//} +// +//func (r attachment) columns() []string { +// return []string{ +// "a.id", +// "a.rel_user", +// "a.url", +// "a.preview_url", +// "a.name", +// "a.meta", +// "a.created_at", +// "a.updated_at", +// "a.deleted_at", +// } +//} +// +//func (r attachment) query() squirrel.SelectBuilder { +// return squirrel. +// Select(r.columns()...). +// From(r.table() + " AS a"). +// Where("a.deleted_at IS NULL") +// +//} +// +//func (r attachment) FindAttachmentByID(ID uint64) (*types.Attachment, error) { +// return r.findOneBy("id", ID) +//} +// +//func (r attachment) findOneBy(field string, value interface{}) (*types.Attachment, error) { +// var ( +// p = &types.Attachment{} +// +// q = r.query(). +// Where(squirrel.Eq{field: value}) +// +// err = rh.FetchOne(r.db(), q, p) +// ) +// +// if err != nil { +// return nil, err +// } else if p.ID == 0 { +// return nil, ErrAttachmentNotFound +// } +// +// return p, nil +//} +// +//func (r attachment) FindAttachmentByMessageID(IDs ...uint64) (rval types.AttachmentSet, err error) { +// rval = types.AttachmentSet{} +// +// if len(IDs) == 0 { +// return +// } +// +// query := r.query(). +// Columns("ma.rel_message"). +// Join(r.tableMessage() + " AS ma ON (a.id = ma.rel_attachment)"). +// Where(squirrel.Eq{"rel_message": IDs}) +// +// return rval, rh.FetchAll(r.db(), query, &rval) +//} +// +//func (r attachment) CreateAttachment(mod *types.Attachment) (*types.Attachment, error) { +// if mod.ID == 0 { +// mod.ID = factory.Sonyflake.NextID() +// } +// +// mod.CreatedAt = time.Now() +// +// return mod, r.db().Insert(r.table(), mod) +//} +// +//func (r attachment) DeleteAttachmentByID(ID uint64) error { +// return rh.UpdateColumns(r.db(), r.table(), rh.Set{"deleted_at": time.Now()}, squirrel.Eq{"id": ID}) +//} +// +//func (r attachment) BindAttachment(attachmentId, messageId uint64) error { +// bond := struct { +// RelAttachment uint64 `db:"rel_attachment"` +// RelMessage uint64 `db:"rel_message"` +// }{attachmentId, messageId} +// +// return r.db().Insert(r.tableMessage(), bond) +//} diff --git a/messaging/repository/error.go b/messaging/repository/error.go index 2c8716406..95d608254 100644 --- a/messaging/repository/error.go +++ b/messaging/repository/error.go @@ -9,9 +9,6 @@ type ( ) const ( - ErrDatabaseError = repositoryError("DatabaseError") - ErrNotImplemented = repositoryError("NotImplemented") - ErrConfigError = repositoryError("ConfigError") ErrEventsPullClosed = repositoryError("EventsPullClosed") ) diff --git a/messaging/repository/mention.go b/messaging/repository/mention.go index 797af5f7b..fa5808d54 100644 --- a/messaging/repository/mention.go +++ b/messaging/repository/mention.go @@ -1,89 +1,89 @@ package repository -import ( - "context" - "time" - - "github.com/Masterminds/squirrel" - "github.com/titpetric/factory" - - "github.com/cortezaproject/corteza-server/messaging/types" - "github.com/cortezaproject/corteza-server/pkg/rh" -) - -type ( - MentionRepository interface { - With(ctx context.Context, db *factory.DB) MentionRepository - - FindByUserIDs(IDs ...uint64) (mm types.MentionSet, err error) - FindByMessageIDs(IDs ...uint64) (mm types.MentionSet, err error) - Create(m *types.Mention) (*types.Mention, error) - DeleteByMessageID(ID uint64) error - DeleteByID(ID uint64) error - } - - mention struct { - *repository - } -) - -var ( - ErrMentionNotFound = repositoryError("MentionNotFound") -) - -func Mention(ctx context.Context, db *factory.DB) MentionRepository { - return (&mention{}).With(ctx, db) -} - -func (r mention) With(ctx context.Context, db *factory.DB) MentionRepository { - return &mention{ - repository: r.repository.With(ctx, db), - } -} - -func (r mention) table() string { - return "messaging_mention" -} - -func (r mention) columns() []string { - return []string{ - "mm.id", - "mm.rel_message", - "mm.rel_channel", - "mm.rel_user", - "mm.rel_mentioned_by", - "mm.created_at", - } -} - -func (r mention) query() squirrel.SelectBuilder { - return squirrel. - Select(r.columns()...). - From(r.table() + " AS mm") -} - -func (r mention) FindByUserIDs(IDs ...uint64) (types.MentionSet, error) { - return r.findAllBy(squirrel.Eq{"rel_user": IDs}) -} - -func (r mention) FindByMessageIDs(IDs ...uint64) (types.MentionSet, error) { - return r.findAllBy(squirrel.Eq{"rel_message": IDs}) -} - -func (r mention) findAllBy(cnd squirrel.Sqlizer) (mm types.MentionSet, err error) { - return mm, rh.FetchAll(r.db(), r.query().Where(cnd), &mm) -} - -func (r mention) Create(m *types.Mention) (*types.Mention, error) { - m.ID = factory.Sonyflake.NextID() - m.CreatedAt = time.Now() - return m, r.db().Insert(r.table(), m) -} - -func (r mention) DeleteByMessageID(ID uint64) error { - return rh.Delete(r.db(), r.table(), squirrel.Eq{"rel_message": ID}) -} - -func (r mention) DeleteByID(ID uint64) error { - return rh.Delete(r.db(), r.table(), squirrel.Eq{"id": ID}) -} +//import ( +// "context" +// "time" +// +// "github.com/Masterminds/squirrel" +// "github.com/titpetric/factory" +// +// "github.com/cortezaproject/corteza-server/messaging/types" +// "github.com/cortezaproject/corteza-server/pkg/rh" +//) +// +//type ( +// MentionRepository interface { +// With(ctx context.Context, db *factory.DB) MentionRepository +// +// FindByUserIDs(IDs ...uint64) (mm types.MentionSet, err error) +// FindByMessageIDs(IDs ...uint64) (mm types.MentionSet, err error) +// Create(m *types.Mention) (*types.Mention, error) +// DeleteByMessageID(ID uint64) error +// DeleteByID(ID uint64) error +// } +// +// mention struct { +// *repository +// } +//) +// +//var ( +// ErrMentionNotFound = repositoryError("MentionNotFound") +//) +// +//func Mention(ctx context.Context, db *factory.DB) MentionRepository { +// return (&mention{}).With(ctx, db) +//} +// +//func (r mention) With(ctx context.Context, db *factory.DB) MentionRepository { +// return &mention{ +// repository: r.repository.With(ctx, db), +// } +//} +// +//func (r mention) table() string { +// return "messaging_mention" +//} +// +//func (r mention) columns() []string { +// return []string{ +// "mm.id", +// "mm.rel_message", +// "mm.rel_channel", +// "mm.rel_user", +// "mm.rel_mentioned_by", +// "mm.created_at", +// } +//} +// +//func (r mention) query() squirrel.SelectBuilder { +// return squirrel. +// Select(r.columns()...). +// From(r.table() + " AS mm") +//} +// +//func (r mention) FindByUserIDs(IDs ...uint64) (types.MentionSet, error) { +// return r.findAllBy(squirrel.Eq{"rel_user": IDs}) +//} +// +//func (r mention) FindByMessageIDs(IDs ...uint64) (types.MentionSet, error) { +// return r.findAllBy(squirrel.Eq{"rel_message": IDs}) +//} +// +//func (r mention) findAllBy(cnd squirrel.Sqlizer) (mm types.MentionSet, err error) { +// return mm, rh.FetchAll(r.db(), r.query().Where(cnd), &mm) +//} +// +//func (r mention) Create(m *types.Mention) (*types.Mention, error) { +// m.ID = factory.Sonyflake.NextID() +// m.CreatedAt = time.Now() +// return m, r.db().Insert(r.table(), m) +//} +// +//func (r mention) DeleteByMessageID(ID uint64) error { +// return rh.Delete(r.db(), r.table(), squirrel.Eq{"rel_message": ID}) +//} +// +//func (r mention) DeleteByID(ID uint64) error { +// return rh.Delete(r.db(), r.table(), squirrel.Eq{"id": ID}) +//} diff --git a/messaging/repository/unread.go b/messaging/repository/unread.go index 5e7209bae..55cd6d5e4 100644 --- a/messaging/repository/unread.go +++ b/messaging/repository/unread.go @@ -32,25 +32,25 @@ type ( const ( sqlResetThreads = `UPDATE messaging_unread - SET count = 0 - WHERE rel_reply_to > 0 AND rel_channel = ? AND rel_user = ?` + SET count = 0 + WHERE rel_reply_to > 0 AND rel_channel = ? AND rel_user = ?` - sqlUnreadIncCount = `UPDATE messaging_unread - SET count = count + 1 - WHERE rel_channel = ? AND rel_reply_to = ? AND rel_user <> ?` + sqlUnreadIncCount = `UPDATE messaging_unread + SET count = count + 1 + WHERE rel_channel = ? AND rel_reply_to = ? AND rel_user <> ?` - sqlUnreadDecCount = `UPDATE messaging_unread - SET count = count - 1 - WHERE rel_channel = ? AND rel_reply_to = ? AND count > 0` + sqlUnreadDecCount = `UPDATE messaging_unread + SET count = count - 1 + WHERE rel_channel = ? AND rel_reply_to = ? AND count > 0` sqlResetCount = `REPLACE INTO messaging_unread (rel_channel, rel_reply_to, rel_user, count) VALUES (?, ?, ?, 0)` sqlUnreadPresetChannel = `INSERT IGNORE INTO messaging_unread (rel_channel, rel_reply_to, rel_user) VALUES (?, ?, ?)` - sqlUnreadPresetThreads = `INSERT IGNORE INTO messaging_unread (rel_channel, rel_reply_to, rel_user) - SELECT rel_channel, id, ? - FROM messaging_message - WHERE rel_channel = ? - AND replies > 0` + sqlUnreadPresetThreads = `INSERT IGNORE INTO messaging_unread (rel_channel, rel_reply_to, rel_user) + SELECT rel_channel, id, ? + FROM messaging_message + WHERE rel_channel = ? + AND replies > 0` ) // Unread creates new instance of channel member repository diff --git a/messaging/rest/attachment.go b/messaging/rest/attachment.go index 11640f49f..254b209d8 100644 --- a/messaging/rest/attachment.go +++ b/messaging/rest/attachment.go @@ -4,14 +4,13 @@ import ( "context" "errors" "fmt" - "io" - "net/http" - "net/url" - - "github.com/cortezaproject/corteza-server/messaging/repository" "github.com/cortezaproject/corteza-server/messaging/rest/request" "github.com/cortezaproject/corteza-server/messaging/service" "github.com/cortezaproject/corteza-server/pkg/auth" + "github.com/cortezaproject/corteza-server/store" + "io" + "net/http" + "net/url" ) type ( @@ -44,7 +43,7 @@ func (ctrl *Attachment) Preview(ctx context.Context, r *request.AttachmentPrevie func (ctrl Attachment) isAccessible(attachmentID, userID uint64, signature string) error { if signature == "" { - return fmt.Errorf("Unauthorized") + return fmt.Errorf("unauthorized") } if userID == 0 { @@ -66,7 +65,7 @@ func (ctrl Attachment) serve(ctx context.Context, ID uint64, preview, download b return func(w http.ResponseWriter, req *http.Request) { att, err := ctrl.att.With(ctx).FindByID(ID) if err != nil { - if errors.Is(err, repository.ErrAttachmentNotFound) { + if errors.Is(err, store.ErrNotFound) { w.WriteHeader(http.StatusNotFound) } else { http.Error(w, err.Error(), http.StatusInternalServerError) diff --git a/messaging/service/attachment.go b/messaging/service/attachment.go index 4c8d5c08b..846e58e96 100644 --- a/messaging/service/attachment.go +++ b/messaging/service/attachment.go @@ -3,24 +3,22 @@ package service import ( "bytes" "context" + "errors" "fmt" + "github.com/cortezaproject/corteza-server/messaging/types" + "github.com/cortezaproject/corteza-server/pkg/actionlog" + intAuth "github.com/cortezaproject/corteza-server/pkg/auth" + "github.com/cortezaproject/corteza-server/pkg/id" + files "github.com/cortezaproject/corteza-server/pkg/store" + "github.com/cortezaproject/corteza-server/store" + "github.com/disintegration/imaging" + "github.com/edwvee/exiffix" "image" "image/gif" "io" "net/http" "path" "strings" - - "github.com/disintegration/imaging" - "github.com/edwvee/exiffix" - - "github.com/titpetric/factory" - - "github.com/cortezaproject/corteza-server/messaging/repository" - "github.com/cortezaproject/corteza-server/messaging/types" - "github.com/cortezaproject/corteza-server/pkg/actionlog" - intAuth "github.com/cortezaproject/corteza-server/pkg/auth" - "github.com/cortezaproject/corteza-server/pkg/store" ) const ( @@ -30,19 +28,15 @@ const ( type ( attachment struct { - db *factory.DB ctx context.Context actionlog actionlog.Recorder ac attachmentAccessController - store store.Store + files files.Store + store store.Storable event EventService - - attachment repository.AttachmentRepository - message repository.MessageRepository - channel repository.ChannelRepository } attachmentAccessController interface { @@ -59,33 +53,29 @@ type ( } ) -func Attachment(ctx context.Context, store store.Store) AttachmentService { +func Attachment(ctx context.Context, store files.Store) AttachmentService { return (&attachment{ ac: DefaultAccessControl, - store: store, + files: store, + store: DefaultNgStore, }).With(ctx) } func (svc attachment) With(ctx context.Context) AttachmentService { - db := repository.DB(ctx) return &attachment{ ctx: ctx, - db: db, ac: svc.ac, actionlog: DefaultActionlog, + files: svc.files, store: svc.store, event: Event(ctx), - - attachment: repository.Attachment(ctx, db), - message: repository.Message(ctx, db), - channel: repository.Channel(ctx, db), } } func (svc attachment) FindByID(id uint64) (*types.Attachment, error) { - return svc.attachment.FindAttachmentByID(id) + return store.LookupMessagingAttachmentByID(svc.ctx, svc.store, id) } func (svc attachment) OpenOriginal(att *types.Attachment) (io.ReadSeeker, error) { @@ -93,7 +83,7 @@ func (svc attachment) OpenOriginal(att *types.Attachment) (io.ReadSeeker, error) return nil, nil } - return svc.store.Open(att.Url) + return svc.files.Open(att.Url) } func (svc attachment) OpenPreview(att *types.Attachment) (io.ReadSeeker, error) { @@ -101,7 +91,7 @@ func (svc attachment) OpenPreview(att *types.Attachment) (io.ReadSeeker, error) return nil, nil } - return svc.store.Open(att.PreviewUrl) + return svc.files.Open(att.PreviewUrl) } func (svc attachment) CreateMessageAttachment(name string, size int64, fh io.ReadSeeker, channelID, replyTo uint64) (att *types.Attachment, err error) { @@ -112,9 +102,10 @@ func (svc attachment) CreateMessageAttachment(name string, size int64, fh io.Rea ch *types.Channel ) - err = svc.db.Transaction(func() (err error) { - if ch, err = svc.channel.FindByID(channelID); err != nil { - if repository.ErrChannelNotFound.Eq(err) { + err = store.Tx(svc.ctx, svc.store, func(ctx context.Context, s store.Storable) (err error) { + + if ch, err = store.LookupMessagingChannelByID(ctx, s, channelID); err != nil { + if errors.Is(err, store.ErrNotFound) { return AttachmentErrChannelNotFound() } } @@ -126,9 +117,10 @@ func (svc attachment) CreateMessageAttachment(name string, size int64, fh io.Rea } att = &types.Attachment{ - ID: factory.Sonyflake.NextID(), - OwnerID: currentUserID, - Name: strings.TrimSpace(name), + ID: id.Next(), + OwnerID: currentUserID, + Name: strings.TrimSpace(name), + CreatedAt: *now(), } err = svc.create(name, size, fh, att) @@ -136,11 +128,12 @@ func (svc attachment) CreateMessageAttachment(name string, size int64, fh io.Rea return err } - if att, err = svc.attachment.CreateAttachment(att); err != nil { + if err = store.CreateMessagingAttachment(ctx, s, att); err != nil { return err } msg := &types.Message{ + ID: id.Next(), Attachment: att, Message: name, Type: types.MessageTypeAttachment, @@ -155,15 +148,17 @@ func (svc attachment) CreateMessageAttachment(name string, size int64, fh io.Rea // Create the first message, doing this directly with repository to circumvent // message service constraints - if msg, err = svc.message.Create(msg); err != nil { + if err = store.CreateMessagingMessage(ctx, s, msg); err != nil { return } aProps.setMessageID(msg.ID) - if err = svc.attachment.BindAttachment(att.ID, msg.ID); err != nil { - return - } + panic("re-implement this") + panic("@todo should we sendEvent?") + //if err = svc.attachment.BindAttachment(att.ID, msg.ID); err != nil { + // return + //} return svc.sendEvent(msg) }) @@ -176,7 +171,7 @@ func (svc attachment) create(name string, size int64, fh io.ReadSeeker, att *typ aProps = &attachmentActionProps{} ) - if svc.store == nil { + if svc.files == nil { return fmt.Errorf("can not create attachment: store handler not set") } @@ -191,10 +186,10 @@ func (svc attachment) create(name string, size int64, fh io.ReadSeeker, att *typ return AttachmentErrFailedToExtractMimeType(aProps).Wrap(err) } - att.Url = svc.store.Original(att.ID, att.Meta.Original.Extension) + att.Url = svc.files.Original(att.ID, att.Meta.Original.Extension) aProps.setUrl(att.Url) - if err = svc.store.Save(att.Url, fh); err != nil { + if err = svc.files.Save(att.Url, fh); err != nil { return AttachmentErrFailedToStoreFile(aProps).Wrap(err) } @@ -317,9 +312,9 @@ func (svc attachment) processImage(original io.ReadSeeker, att *types.Attachment meta.Extension = f2e[previewFormat] // Can and how we make a preview of this attachment? - att.PreviewUrl = svc.store.Preview(att.ID, meta.Extension) + att.PreviewUrl = svc.files.Preview(att.ID, meta.Extension) - return svc.store.Save(att.PreviewUrl, buf) + return svc.files.Save(att.PreviewUrl, buf) } // Sends message to event loop diff --git a/messaging/service/channel.go b/messaging/service/channel.go index 60dafd5b6..edba989d0 100644 --- a/messaging/service/channel.go +++ b/messaging/service/channel.go @@ -521,7 +521,7 @@ func (svc *channel) Delete(ID uint64) (ch *types.Channel, err error) { return } else { // Set deletedAt timestamp so that our clients can react properly... - ch.DeletedAt = timeNowPtr() + ch.DeletedAt = now() } _ = svc.sendChannelEvent(ch) @@ -648,7 +648,7 @@ func (svc *channel) Archive(ID uint64) (ch *types.Channel, err error) { return } else { // Set archivedAt timestamp so that our clients can react properly... - ch.ArchivedAt = timeNowPtr() + ch.ArchivedAt = now() } _ = svc.flushSystemMessages() diff --git a/messaging/service/message.go b/messaging/service/message.go index 994b4dde0..9e4b9e7f0 100644 --- a/messaging/service/message.go +++ b/messaging/service/message.go @@ -2,34 +2,25 @@ package service import ( "context" + "errors" + "fmt" + "github.com/cortezaproject/corteza-server/messaging/types" + "github.com/cortezaproject/corteza-server/pkg/auth" + "github.com/cortezaproject/corteza-server/pkg/id" + "github.com/cortezaproject/corteza-server/pkg/payload" + "github.com/cortezaproject/corteza-server/store" "io" "regexp" "strings" - - "github.com/pkg/errors" - - "github.com/cortezaproject/corteza-server/messaging/repository" - "github.com/cortezaproject/corteza-server/messaging/types" - "github.com/cortezaproject/corteza-server/pkg/auth" - "github.com/cortezaproject/corteza-server/pkg/payload" ) type ( message struct { - db db - ctx context.Context - ac messageAccessController - + ctx context.Context + ac messageAccessController channel ChannelService - - attachment repository.AttachmentRepository - cmember repository.ChannelMemberRepository - unread repository.UnreadRepository - message repository.MessageRepository - mflag repository.MessageFlagRepository - mentions repository.MentionRepository - - event EventService + store store.Storable + event EventService } messageAccessController interface { @@ -80,25 +71,17 @@ func Message(ctx context.Context) MessageService { return (&message{ ac: DefaultAccessControl, channel: DefaultChannel, + store: DefaultNgStore, }).With(ctx) } func (svc message) With(ctx context.Context) MessageService { - db := repository.DB(ctx) return &message{ - db: db, ctx: ctx, ac: svc.ac, channel: svc.channel, - - event: Event(ctx), - - attachment: repository.Attachment(ctx, db), - cmember: repository.ChannelMember(ctx, db), - unread: repository.Unread(ctx, db), - message: repository.Message(ctx, db), - mflag: repository.MessageFlag(ctx, db), - mentions: repository.Mention(ctx, db), + store: svc.store, + event: Event(ctx), } } @@ -109,12 +92,12 @@ func (svc message) Find(filter types.MessageFilter) (mm types.MessageSet, f type return } - mm, f, err = svc.message.Find(f) + mm, f, err = store.SearchMessagingMessages(svc.ctx, svc.store, f) if err != nil { return } - return mm, f, svc.preload(mm) + return mm, f, svc.preload(svc.ctx, svc.store, mm) } func (svc message) FindThreads(filter types.MessageFilter) (mm types.MessageSet, f types.MessageFilter, err error) { @@ -124,12 +107,12 @@ func (svc message) FindThreads(filter types.MessageFilter) (mm types.MessageSet, return } - mm, f, err = svc.message.FindThreads(f) + mm, f, err = store.SearchMessagingThreads(svc.ctx, svc.store, f) if err != nil { return } - return mm, f, svc.preload(mm) + return mm, f, svc.preload(svc.ctx, svc.store, mm) } func (svc message) CreateWithAvatar(in *types.Message, avatar io.Reader) (*types.Message, error) { @@ -159,40 +142,34 @@ func (svc message) readableChannels(f types.MessageFilter) ([]uint64, error) { return cc.IDs(), nil } -func (svc message) Create(in *types.Message) (m *types.Message, err error) { - if in == nil { - in = &types.Message{} - } +func (svc message) Create(msg *types.Message) (*types.Message, error) { + msg.ID = id.Next() + msg.CreatedAt = *now() + msg.Message = strings.TrimSpace(msg.Message) - in.Message = strings.TrimSpace(in.Message) - - var mlen = len(in.Message) - - if mlen == 0 { - return nil, errors.Errorf("refusing to create message without contents") - } - - if settingsMessageBodyLength > 0 && mlen > settingsMessageBodyLength { - return nil, errors.Errorf("message length (%d characters) too long (max: %d)", mlen, settingsMessageBodyLength) + if l := len(msg.Message); l == 0 { + return nil, fmt.Errorf("refusing to create message without contents") + } else if settingsMessageBodyLength > 0 && l > settingsMessageBodyLength { + return nil, fmt.Errorf("message length (%d characters) too long (max: %d)", l, settingsMessageBodyLength) } // keep pre-existing user id set - if in.UserID == 0 { - in.UserID = auth.GetIdentityFromContext(svc.ctx).Identity() + if msg.UserID == 0 { + msg.UserID = auth.GetIdentityFromContext(svc.ctx).Identity() } - return m, svc.db.Transaction(func() (err error) { + err := store.Tx(svc.ctx, svc.store, func(ctx context.Context, s store.Storable) (err error) { // Broadcast queue var bq = types.MessageSet{} var ch *types.Channel - if in.ReplyTo > 0 { + if msg.ReplyTo > 0 { var original *types.Message - var replyTo = in.ReplyTo + var replyTo = msg.ReplyTo for replyTo > 0 { // Find original message - original, err = svc.message.FindByID(in.ReplyTo) + original, err = store.LookupMessagingMessageByID(ctx, s, msg.ReplyTo) if err != nil { return } @@ -201,34 +178,35 @@ func (svc message) Create(in *types.Message) (m *types.Message, err error) { } if !original.Type.IsRepliable() { - return errors.Errorf("unable to reply on this message (type = %s)", original.Type) + return fmt.Errorf("unable to reply on this message (type = %s)", original.Type) } // We do not want to have multi-level threads // Take original's reply-to and use it - in.ReplyTo = original.ID + msg.ReplyTo = original.ID - in.ChannelID = original.ChannelID + msg.ChannelID = original.ChannelID if original.Replies == 0 { // First reply, // // reset unreads for all members - var mm types.ChannelMemberSet - mm, err = svc.cmember.Find(types.ChannelMemberFilterChannels(original.ChannelID)) - if err != nil { - return err - } + //var mm types.ChannelMemberSet + //mm, _, err = store.SearchMessagingChannelMembers(svc.ctx, svc.store, types.ChannelMemberFilterChannels(original.ChannelID)) + //if err != nil { + // return err + //} - err = svc.unread.Preset(original.ChannelID, original.ID, mm.AllMemberIDs()...) - if err != nil { - return err - } + panic("reimplement this") + //err = svc.unread.Preset(original.ChannelID, original.ID, mm.AllMemberIDs()...) + //if err != nil { + // return err + //} } - // Increment counter, on struct and in repository. + // Increment counter, on struct and in store. original.Replies++ - if err = svc.message.IncReplyCount(original.ID); err != nil { + if err = store.PartialMessagingMessageUpdate(ctx, s, []string{"replies"}, original); err != nil { return } @@ -236,115 +214,126 @@ func (svc message) Create(in *types.Message) (m *types.Message, err error) { bq = append(bq, original) } - if in.ChannelID == 0 { - return errors.New("channelID missing") - } else if ch, err = svc.findChannelByID(in.ChannelID); err != nil { + if msg.ChannelID == 0 { + return fmt.Errorf("channelID missing") + } else if ch, err = svc.findChannelByID(msg.ChannelID); err != nil { return } - if in.ReplyTo > 0 && !svc.ac.CanReplyMessage(svc.ctx, ch) { + if msg.ReplyTo > 0 && !svc.ac.CanReplyMessage(ctx, ch) { return ErrNoPermissions.withStack() } if !svc.ac.CanSendMessage(svc.ctx, ch) { return ErrNoPermissions.withStack() } - if m, err = svc.message.Create(in); err != nil { + if err = store.CreateMessagingMessage(ctx, s, msg); err != nil { return } - mentions := svc.extractMentions(m) - if err = svc.updateMentions(m.ID, mentions); err != nil { + mentions := svc.extractMentions(msg) + if err = svc.updateMentions(ctx, s, msg.ID, mentions); err != nil { return } - svc.sendNotifications(m, mentions) + svc.sendNotifications(msg, mentions) // Count unreads in the background and send updates to all users - svc.countUnreads(ch, m, 0) + if err = svc.countUnreads(ctx, s, ch, msg, 0); err != nil { + return + } - return svc.sendEvent(append(bq, m)...) + return svc.sendEvent(append(bq, msg)...) }) + + if err != nil { + return nil, err + } + + return msg, err } -func (svc message) Update(in *types.Message) (message *types.Message, err error) { - if in.ID == 0 { +func (svc message) Update(upd *types.Message) (msg *types.Message, err error) { + if upd.ID == 0 { return nil, ErrInvalidID.withStack() } - if in == nil { - in = &types.Message{} - } + upd.Message = strings.TrimSpace(upd.Message) - in.Message = strings.TrimSpace(in.Message) - var mlen = len(in.Message) - - if mlen == 0 { - return nil, errors.Errorf("refusing to update message without contents") - } - if settingsMessageBodyLength > 0 && mlen > settingsMessageBodyLength { - return nil, errors.Errorf("message length (%d characters) too long (max: %d)", mlen, settingsMessageBodyLength) + if l := len(upd.Message); l == 0 { + return nil, fmt.Errorf("refusing to update message without contents") + } else if settingsMessageBodyLength > 0 && l > settingsMessageBodyLength { + return nil, fmt.Errorf("message length (%d characters) too long (max: %d)", l, settingsMessageBodyLength) } var currentUserID = auth.GetIdentityFromContext(svc.ctx).Identity() - return message, svc.db.Transaction(func() (err error) { + err = store.Tx(svc.ctx, svc.store, func(ctx context.Context, s store.Storable) (err error) { var ch *types.Channel - if ch, err = svc.findChannelByID(in.ChannelID); err != nil { + if ch, err = svc.findChannelByID(upd.ChannelID); err != nil { return err - } else if !svc.ac.CanReadChannel(svc.ctx, ch) { + } else if !svc.ac.CanReadChannel(ctx, ch) { return ErrNoPermissions.withStack() } - message, err = svc.message.FindByID(in.ID) + msg, err = store.LookupMessagingMessageByID(ctx, s, upd.ID) if err != nil { - return errors.Wrap(err, "could not load message for editing") + return fmt.Errorf("could not load message for editing: %w", err) } - if message.Message == in.Message { + if msg.Message == upd.Message { // Nothing changed return nil } - if message.UserID == currentUserID && !svc.ac.CanUpdateOwnMessages(svc.ctx, ch) { + if msg.UserID == currentUserID && !svc.ac.CanUpdateOwnMessages(ctx, ch) { return ErrNoPermissions.withStack() - } else if message.UserID != currentUserID && !svc.ac.CanUpdateMessages(svc.ctx, ch) { + } else if msg.UserID != currentUserID && !svc.ac.CanUpdateMessages(ctx, ch) { return ErrNoPermissions.withStack() } - // Allow message content to be changed - message.Message = in.Message + // Update message + msg.Message = upd.Message + msg.UpdatedAt = now() - if message, err = svc.message.Update(message); err != nil { + if err = store.UpdateMessagingMessage(ctx, s, msg); err != nil { return err } - if err = svc.updateMentions(message.ID, svc.extractMentions(message)); err != nil { + if err = svc.updateMentions(ctx, s, msg.ID, svc.extractMentions(msg)); err != nil { return } - return svc.sendEvent(message) + return svc.sendEvent(msg) }) + + if err != nil { + return nil, err + } + + return msg, err } -func (svc message) Delete(messageID uint64) error { +func (svc message) Delete(messageID uint64) (err error) { var currentUserID = auth.GetIdentityFromContext(svc.ctx).Identity() _ = currentUserID - return svc.db.Transaction(func() (err error) { + err = store.Tx(svc.ctx, svc.store, func(ctx context.Context, s store.Storable) (err error) { // Broadcast queue var bq = types.MessageSet{} var deletedMsg, original *types.Message var ch *types.Channel - deletedMsg, err = svc.message.FindByID(messageID) + deletedMsg, err = store.LookupMessagingMessageByID(ctx, s, messageID) if err != nil { return err } + deletedMsg.DeletedAt = now() + if ch, err = svc.findChannelByID(deletedMsg.ChannelID); err != nil { return err } else if !svc.ac.CanReadChannel(svc.ctx, ch) { @@ -358,7 +347,7 @@ func (svc message) Delete(messageID uint64) error { } if deletedMsg.ReplyTo > 0 { - original, err = svc.message.FindByID(deletedMsg.ReplyTo) + original, err = store.LookupMessagingMessageByID(ctx, s, deletedMsg.ReplyTo) if err != nil { return err } @@ -369,30 +358,34 @@ func (svc message) Delete(messageID uint64) error { original.Replies-- } - if err = svc.message.DecReplyCount(original.ID); err != nil { - return err + if err = store.PartialMessagingMessageUpdate(ctx, s, []string{"replies"}, original); err != nil { + return } // Broadcast updated original bq = append(bq, original) } - if err = svc.message.DeleteByID(messageID); err != nil { + if err = store.UpdateMessagingMessage(ctx, s, deletedMsg); err != nil { return } // Set deletedAt timestamp so that our clients can react properly... - deletedMsg.DeletedAt = timeNowPtr() + deletedMsg.DeletedAt = now() - if err = svc.updateMentions(messageID, nil); err != nil { + if err = svc.updateMentions(ctx, s, messageID, nil); err != nil { return } // Count unreads in the background and send updates to all users - svc.countUnreads(ch, deletedMsg, 0) + if err = svc.countUnreads(ctx, s, ch, deletedMsg, 0); err != nil { + return + } return svc.sendEvent(append(bq, deletedMsg)...) }) + + return } // MarkAsRead marks channel/thread as read @@ -400,13 +393,14 @@ func (svc message) Delete(messageID uint64) error { // If lastReadMessageID is set, it uses that message as last read message func (svc message) MarkAsRead(channelID, threadID, lastReadMessageID uint64) (uint64, uint32, uint32, error) { var ( - currentUserID uint64 = auth.GetIdentityFromContext(svc.ctx).Identity() - count uint32 - threadCount uint32 - err error + currentUserID = auth.GetIdentityFromContext(svc.ctx).Identity() + + count uint32 + threadCount uint32 + err error ) - err = svc.db.Transaction(func() (err error) { + err = store.Tx(svc.ctx, svc.store, func(ctx context.Context, s store.Storable) (err error) { var ch *types.Channel var thread *types.Message var lastMessage *types.Message @@ -416,21 +410,21 @@ func (svc message) MarkAsRead(channelID, threadID, lastReadMessageID uint64) (ui } else if !svc.ac.CanReadChannel(svc.ctx, ch) { return ErrNoPermissions.withStack() } else if !ch.IsValid() { - return errors.New("invalid channel") + return fmt.Errorf("invalid channel") } if threadID > 0 { // Validate thread - if thread, err = svc.message.FindByID(threadID); err != nil { - return errors.Wrap(err, "unable to verify thread") + if thread, err = store.LookupMessagingMessageByID(ctx, s, threadID); err != nil { + return fmt.Errorf("unable to verify thread: %w", err) } else if !thread.IsValid() { - return errors.New("invalid thread") + return fmt.Errorf("invalid thread") } } else { // This is request for channel, // count all thread unreads var uu types.UnreadSet - uu, err = svc.unread.CountThreads(currentUserID, channelID) + uu, err = store.CountMessagingUnreadThreads(ctx, s, currentUserID, channelID) if err != nil { return err } @@ -442,51 +436,59 @@ func (svc message) MarkAsRead(channelID, threadID, lastReadMessageID uint64) (ui if lastReadMessageID > 0 { // Validate messageID/threadID/channelID combo - if lastMessage, err = svc.message.FindByID(lastReadMessageID); err != nil { - return errors.Wrap(err, "unable to verify last message") + if lastMessage, err = store.LookupMessagingMessageByID(ctx, s, lastReadMessageID); err != nil { + return fmt.Errorf("unable to verify last message: %w", err) } else if !lastMessage.IsValid() { - return errors.New("invalid message") + return fmt.Errorf("invalid message") } else if lastMessage.ChannelID != channelID { - return errors.New("last read message not in the same channel") + return fmt.Errorf("last read message not in the same channel") } else if threadID > 0 && lastMessage.ReplyTo != threadID { - return errors.New("last read message not in the same thread") + return fmt.Errorf("last read message not in the same thread") } - count, err = svc.message.CountFromMessageID(channelID, threadID, lastReadMessageID) + count, err = store.CountMessagingMessagesFromID(ctx, s, channelID, threadID, lastReadMessageID) if err != nil { - return errors.Wrap(err, "unable to count unread messages") + return fmt.Errorf("unable to count unread messages: %w", err) } } else { // use last message ID - if lastReadMessageID, err = svc.message.LastMessageID(channelID, threadID); err != nil { - return errors.Wrap(err, "unable to find last message") + if lastReadMessageID, err = store.LastMessagingMessageID(ctx, s, channelID, threadID); err != nil { + return fmt.Errorf("unable to find last message: %w", err) } // no need to count count = 0 } - err = svc.unread.Record(currentUserID, channelID, threadID, lastReadMessageID, count) - if err != nil { - return errors.Wrap(err, "unable to record unread messages") + u := &types.Unread{ + ChannelID: channelID, + UserID: currentUserID, + ReplyTo: threadID, + LastMessageID: lastReadMessageID, + Count: count, + } + + if err = store.UpsertMessagingUnread(ctx, s, u); err != nil { + return fmt.Errorf("unable to record unread messages: %w", err) } // Remove unread counts from all threads when doing mark-channel-as-read if threadID == 0 { - err = svc.unread.ClearThreads(channelID, currentUserID) - if err != nil { - return errors.Wrap(err, "unable to clear channel threads") + if err = store.ResetMessagingUnreadThreads(ctx, s, channelID, currentUserID); err != nil { + return fmt.Errorf("unable to clear channel threads: %w", err) } } // Re-count unreads and send updates to this user - svc.countUnreads(ch, nil, currentUserID) + if err = svc.countUnreads(ctx, s, ch, nil, currentUserID); err != nil { + return + } return nil }) - return lastReadMessageID, count, threadCount, errors.Wrap(err, "unable to mark as read") + return lastReadMessageID, count, threadCount, fmt.Errorf("unable to mark as read: %w", err) } // React on a message with an emoji @@ -520,7 +522,7 @@ func (svc message) RemoveBookmark(messageID uint64) error { } // React on a message with an emoji -func (svc message) flag(messageID uint64, flag string, remove bool) error { +func (svc message) flag(messageID uint64, flag string, remove bool) (err error) { var currentUserID = auth.GetIdentityFromContext(svc.ctx).Identity() _ = currentUserID @@ -530,20 +532,20 @@ func (svc message) flag(messageID uint64, flag string, remove bool) error { flag = types.MessageFlagPinnedToChannel } - // @todo validate flags beyond empty string + // @todo validate flags more strictly - err := svc.db.Transaction(func() (err error) { - var flagOwnerId = currentUserID + err = store.Tx(svc.ctx, svc.store, func(ctx context.Context, s store.Storable) (err error) { + var flagOwnerID = currentUserID var f *types.MessageFlag var msg *types.Message var ch *types.Channel if flag == types.MessageFlagPinnedToChannel { // It does not matter how is the owner of the pin, - flagOwnerId = 0 + flagOwnerID = 0 } - f, err = svc.mflag.FindByFlag(messageID, flagOwnerId, flag) + f, err = store.LookupMessagingFlagByMessageIDUserIDFlag(ctx, s, messageID, flagOwnerID, flag) if f == nil && remove { // Skip removing, flag does not exists return nil @@ -554,12 +556,12 @@ func (svc message) flag(messageID uint64, flag string, remove bool) error { return nil } - if err != nil && err != repository.ErrMessageFlagNotFound { + if err != nil && errors.Is(err, store.ErrNotFound) { // Other errors, exit return } - if msg, err = svc.message.FindByID(messageID); err != nil { + if msg, err = store.LookupMessagingMessageByID(svc.ctx, svc.store, messageID); err != nil { return } @@ -576,15 +578,18 @@ func (svc message) flag(messageID uint64, flag string, remove bool) error { } if remove { - err = svc.mflag.DeleteByID(f.ID) - f.DeletedAt = timeNowPtr() + f.DeletedAt = now() + err = store.UpdateMessagingFlag(ctx, s, f) } else { - f, err = svc.mflag.Create(&types.MessageFlag{ + f = &types.MessageFlag{ + ID: id.Next(), UserID: currentUserID, + CreatedAt: *now(), ChannelID: msg.ChannelID, MessageID: msg.ID, Flag: flag, - }) + } + err = store.CreateMessagingFlag(ctx, s, f) } if err != nil { @@ -595,27 +600,27 @@ func (svc message) flag(messageID uint64, flag string, remove bool) error { return }) - return errors.Wrap(err, "can not flag/un-flag message") + return fmt.Errorf("can not flag: %w/un-flag message", err) } -func (svc message) preload(mm types.MessageSet) (err error) { - if err = svc.preloadAttachments(mm); err != nil { +func (svc message) preload(ctx context.Context, s store.Storable, mm types.MessageSet) (err error) { + if err = svc.preloadAttachments(ctx, s, mm); err != nil { return } - if err = svc.preloadFlags(mm); err != nil { + if err = svc.preloadFlags(ctx, s, mm); err != nil { return } - if err = svc.preloadMentions(mm); err != nil { + if err = svc.preloadMentions(ctx, s, mm); err != nil { return } - if err = svc.preloadUnreads(mm); err != nil { + if err = svc.preloadUnreads(ctx, s, mm); err != nil { return } - if err = svc.message.PrefillThreadParticipants(mm); err != nil { + if err = svc.preloadThreadParticipants(ctx, s, mm); err != nil { return } @@ -623,10 +628,13 @@ func (svc message) preload(mm types.MessageSet) (err error) { } // Preload for all messages -func (svc message) preloadFlags(mm types.MessageSet) (err error) { - var ff types.MessageFlagSet +func (message) preloadFlags(ctx context.Context, s store.Storable, mm types.MessageSet) (err error) { + if len(mm) == 0 { + return + } - ff, err = svc.mflag.FindByMessageIDs(mm.IDs()...) + var ff types.MessageFlagSet + ff, _, err = store.SearchMessagingFlags(ctx, s, types.MessageFlagFilter{MessageID: mm.IDs()}) if err != nil { return } @@ -638,10 +646,13 @@ func (svc message) preloadFlags(mm types.MessageSet) (err error) { } // Preload for all messages -func (svc message) preloadMentions(mm types.MessageSet) (err error) { - var mentions types.MentionSet +func (message) preloadMentions(ctx context.Context, s store.Storable, mm types.MessageSet) (err error) { + if len(mm) == 0 { + return + } - mentions, err = svc.mentions.FindByMessageIDs(mm.IDs()...) + var mentions types.MentionSet + mentions, _, err = store.SearchMessagingMentions(ctx, s, types.MentionFilter{MessageID: mm.IDs()}) if err != nil { return } @@ -652,24 +663,16 @@ func (svc message) preloadMentions(mm types.MessageSet) (err error) { }) } -func (svc message) preloadAttachments(mm types.MessageSet) (err error) { +func (message) preloadAttachments(ctx context.Context, s store.Storable, mm types.MessageSet) (err error) { var ( - ids []uint64 - aa types.AttachmentSet + aa types.AttachmentSet ) - err = mm.Walk(func(m *types.Message) error { - if m.Type == types.MessageTypeAttachment || m.Type == types.MessageTypeInlineImage { - ids = append(ids, m.ID) - } - return nil - }) - - if err != nil { + if err != nil || len(mm) == 0 { return } - if aa, err = svc.attachment.FindAttachmentByMessageID(ids...); err != nil { + if aa, _, err = store.SearchMessagingAttachments(ctx, s, types.AttachmentFilter{MessageID: mm.IDs()}); err != nil { return } else { _ = aa @@ -686,31 +689,61 @@ func (svc message) preloadAttachments(mm types.MessageSet) (err error) { } } -func (svc message) preloadUnreads(mm types.MessageSet) error { - var userID = auth.GetIdentityFromContext(svc.ctx).Identity() +func (message) preloadUnreads(ctx context.Context, s store.Storable, mm types.MessageSet) (err error) { + if len(mm) == 0 { + return nil + } + + var ( + unread types.UnreadSet + userID = auth.GetIdentityFromContext(ctx).Identity() + ) // Filter out only relevant messages -- ones with replies mm, _ = mm.Filter(func(m *types.Message) (b bool, e error) { return m.Replies > 0, nil }) + unread, err = store.CountMessagingUnread(ctx, s, userID, 0, mm.IDs()...) + if err != nil { + return + } + + return mm.Walk(func(m *types.Message) error { + m.Unread = unread.FindByThreadId(m.ID) + return nil + }) +} + +func (message) preloadThreadParticipants(ctx context.Context, s store.Storable, mm types.MessageSet) (err error) { if len(mm) == 0 { return nil } - if vv, err := svc.unread.Count(userID, 0, mm.IDs()...); err != nil { - return err - } else { - return mm.Walk(func(m *types.Message) error { - m.Unread = vv.FindByThreadId(m.ID) - return nil - }) + var ( + unread types.UnreadSet + userID = auth.GetIdentityFromContext(ctx).Identity() + ) + + // Filter out only relevant messages -- ones with replies + mm, _ = mm.Filter(func(m *types.Message) (b bool, e error) { + return m.Replies > 0, nil + }) + + unread, err = store.CountMessagingUnread(ctx, s, userID, 0, mm.IDs()...) + if err != nil { + return } + + return mm.Walk(func(m *types.Message) error { + m.Unread = unread.FindByThreadId(m.ID) + return nil + }) } // Sends message to event loop func (svc message) sendEvent(mm ...*types.Message) (err error) { - if err = svc.preload(mm); err != nil { + if err = svc.preload(svc.ctx, svc.store, mm); err != nil { return } @@ -735,9 +768,8 @@ func (svc message) sendNotifications(message *types.Message, mentions types.Ment // 1. increases/decreases unread counters for channel or thread // 2. collects all counters for channel or thread // 3. sends unread events to subscribers -func (svc message) countUnreads(ch *types.Channel, m *types.Message, userID uint64) { +func (svc message) countUnreads(ctx context.Context, s store.Storable, ch *types.Channel, m *types.Message, userID uint64) (err error) { var ( - err error uuBase, uuThreads, uuChannels types.UnreadSet // mm types.ChannelMemberSet threadIDs []uint64 @@ -746,21 +778,25 @@ func (svc message) countUnreads(ch *types.Channel, m *types.Message, userID uint if m != nil { if m.DeletedAt != nil { // When deleting message, all existing counters are decreased! - if err = svc.unread.Dec(m.ChannelID, m.ReplyTo, m.UserID); err != nil { + if err = store.DecMessagingUnreadCount(ctx, s, m.ChannelID, m.ReplyTo, m.UserID); err != nil { return } } else if m.UpdatedAt == nil { // Reset user's counter and set current message ID as last read. - err = svc.unread.Record( - m.UserID, - m.ChannelID, - m.ReplyTo, - m.ID, - 0, - ) + u := &types.Unread{ + ChannelID: m.ChannelID, + UserID: m.UserID, + ReplyTo: m.ReplyTo, + LastMessageID: m.ID, + Count: 0, + } + + if err = store.UpsertMessagingUnread(ctx, s, u); err != nil { + return err + } // When new message is created, update all existing counters - if err = svc.unread.Inc(m.ChannelID, m.ReplyTo, m.UserID); err != nil { + if err = store.IncMessagingUnreadCount(ctx, s, m.ChannelID, m.ReplyTo, m.UserID); err != nil { return } } @@ -770,7 +806,7 @@ func (svc message) countUnreads(ch *types.Channel, m *types.Message, userID uint } } - uuBase, err = svc.unread.Count(userID, ch.ID, threadIDs...) + uuBase, err = store.CountMessagingUnread(ctx, s, userID, ch.ID, threadIDs...) if err != nil { return } @@ -778,7 +814,7 @@ func (svc message) countUnreads(ch *types.Channel, m *types.Message, userID uint if len(threadIDs) > 0 { // If base count was done for a thread, // Do another count for channel - uuChannels, err = svc.unread.Count(userID, ch.ID) + uuChannels, err = store.CountMessagingUnread(ctx, s, userID, ch.ID) if err != nil { return } @@ -786,7 +822,7 @@ func (svc message) countUnreads(ch *types.Channel, m *types.Message, userID uint uuBase = uuBase.Merge(uuChannels) // Now recount all threads for this channel - uuThreads, err = svc.unread.CountThreads(userID, ch.ID) + uuThreads, err = store.CountMessagingUnreadThreads(ctx, s, userID, ch.ID) if err != nil { return } @@ -795,10 +831,7 @@ func (svc message) countUnreads(ch *types.Channel, m *types.Message, userID uint } // This is a reply, make sure we fetch the new stats about unread replies and push them to users - err = svc.event.UnreadCounters(uuBase) - if err != nil { - return - } + return svc.event.UnreadCounters(uuBase) } // Sends message to event loop @@ -838,30 +871,21 @@ func (svc message) extractMentions(m *types.Message) (mm types.MentionSet) { return } -func (svc message) updateMentions(messageID uint64, mm types.MentionSet) error { - if existing, err := svc.mentions.FindByMessageIDs(messageID); err != nil { - return errors.Wrap(err, "could not update mentions") +func (svc message) updateMentions(ctx context.Context, s store.Storable, messageID uint64, mm types.MentionSet) error { + if existing, _, err := store.SearchMessagingMentions(ctx, s, types.MentionFilter{MessageID: []uint64{messageID}}); err != nil { + return fmt.Errorf("could not update mentions: %w", err) } else if len(mm) > 0 { add, _, del := existing.Diff(mm) - err = add.Walk(func(m *types.Mention) error { - _, err = svc.mentions.Create(m) - return err - }) - - if err != nil { - return errors.Wrap(err, "could not create mentions") + if err = store.CreateMessagingMention(ctx, s, add...); err != nil { + return fmt.Errorf("could not create mentions: %w", err) } - err = del.Walk(func(m *types.Mention) error { - return svc.mentions.DeleteByID(m.ID) - }) - - if err != nil { - return errors.Wrap(err, "could not delete mentions") + if err = store.DeleteMessagingMention(ctx, s, del...); err != nil { + return fmt.Errorf("could not delete mentions: %w", err) } } else { - return svc.mentions.DeleteByMessageID(messageID) + return store.DeleteMessagingMentionByID(svc.ctx, svc.store, messageID) } return nil @@ -876,7 +900,7 @@ func (svc message) findChannelByID(channelID uint64) (ch *types.Channel, err err if ch, err = svc.channel.With(svc.ctx).FindByID(channelID); err != nil { return nil, err - } else if mm, err = svc.cmember.Find(types.ChannelMemberFilterChannels(ch.ID)); err != nil { + } else if mm, _, err = store.SearchMessagingChannelMembers(nil, nil, types.ChannelMemberFilterChannels(ch.ID)); err != nil { return nil, err } else { ch.Members = mm.AllMemberIDs() diff --git a/messaging/service/service.go b/messaging/service/service.go index 8fdda4847..2bf4ce7f4 100644 --- a/messaging/service/service.go +++ b/messaging/service/service.go @@ -141,7 +141,7 @@ func Watchers(ctx context.Context) { DefaultPermissions.Watch(ctx) } -func timeNowPtr() *time.Time { +func now() *time.Time { now := time.Now() return &now } diff --git a/messaging/types/mention.go b/messaging/types/mention.go index 56d37f028..7df55a645 100644 --- a/messaging/types/mention.go +++ b/messaging/types/mention.go @@ -6,22 +6,23 @@ import ( type ( Mention struct { - ID uint64 `db:"id"` - MessageID uint64 `db:"rel_message"` - ChannelID uint64 `db:"rel_channel"` - UserID uint64 `db:"rel_user"` - MentionedByID uint64 `db:"rel_mentioned_by"` - CreatedAt time.Time `db:"created_at"` + ID uint64 + MessageID uint64 + ChannelID uint64 + UserID uint64 + MentionedByID uint64 + CreatedAt time.Time } MentionFilter struct { - // All mentions by this user - MentionedByID uint64 - - // All mentions of this user - UserID uint64 - - // How many entries - Limit uint + MessageID []uint64 + //// All mentions by this user + //MentionedByID []uint64 + // + //// All mentions of this user + //UserID []uint64 + // + //// How many entries + //Limit uint } ) diff --git a/messaging/types/message_flag.go b/messaging/types/message_flag.go index 2654abca0..651e7b5fa 100644 --- a/messaging/types/message_flag.go +++ b/messaging/types/message_flag.go @@ -6,15 +6,20 @@ import ( type ( MessageFlag struct { - ID uint64 `json:"id" db:"id"` - UserID uint64 `json:"userId" db:"rel_user"` - MessageID uint64 `json:"messageId" db:"rel_message"` - ChannelID uint64 `json:"channelId" db:"rel_channel"` - Flag string `json:"flag" db:"flag"` - CreatedAt time.Time `json:"createdAt,omitempty" db:"created_at"` + ID uint64 `json:"id"` + UserID uint64 `json:"userId"` + MessageID uint64 `json:"messageId"` + ChannelID uint64 `json:"channelId"` + Flag string `json:"flag"` + CreatedAt time.Time `json:"createdAt,omitempty"` // Internal only - DeletedAt *time.Time `json:"-" db:"-"` + DeletedAt *time.Time `json:"-"` + } + + MessageFlagFilter struct { + Flag string + MessageID []uint64 } ) diff --git a/store/interfaces.gen.go b/store/interfaces.gen.go index ff3d41ff0..2c9066a1d 100644 --- a/store/interfaces.gen.go +++ b/store/interfaces.gen.go @@ -19,6 +19,8 @@ package store // - store/messaging_attachments.yaml // - store/messaging_channel_members.yaml // - store/messaging_channels.yaml +// - store/messaging_flags.yaml +// - store/messaging_mentions.yaml // - store/messaging_messages.yaml // - store/messaging_unread.yaml // - store/rbac_rules.yaml @@ -60,6 +62,8 @@ type ( MessagingAttachments MessagingChannelMembers MessagingChannels + MessagingFlags + MessagingMentions MessagingMessages MessagingUnreads RbacRules diff --git a/store/messaging_flags.gen.go b/store/messaging_flags.gen.go new file mode 100644 index 000000000..1d61039d2 --- /dev/null +++ b/store/messaging_flags.gen.go @@ -0,0 +1,87 @@ +package store + +// This file is auto-generated. +// +// Template: pkg/codegen/assets/store_base.gen.go.tpl +// Definitions: store/messaging_flags.yaml +// +// Changes to this file may cause incorrect behavior and will be lost if +// the code is regenerated. + +import ( + "context" + "github.com/cortezaproject/corteza-server/messaging/types" +) + +type ( + MessagingFlags interface { + SearchMessagingFlags(ctx context.Context, f types.MessageFlagFilter) (types.MessageFlagSet, types.MessageFlagFilter, error) + LookupMessagingFlagByID(ctx context.Context, id uint64) (*types.MessageFlag, error) + LookupMessagingFlagByMessageIDUserIDFlag(ctx context.Context, message_id uint64, user_id uint64, flag string) (*types.MessageFlag, error) + + CreateMessagingFlag(ctx context.Context, rr ...*types.MessageFlag) error + + UpdateMessagingFlag(ctx context.Context, rr ...*types.MessageFlag) error + PartialMessagingFlagUpdate(ctx context.Context, onlyColumns []string, rr ...*types.MessageFlag) error + + UpsertMessagingFlag(ctx context.Context, rr ...*types.MessageFlag) error + + DeleteMessagingFlag(ctx context.Context, rr ...*types.MessageFlag) error + DeleteMessagingFlagByID(ctx context.Context, ID uint64) error + + TruncateMessagingFlags(ctx context.Context) error + } +) + +var _ *types.MessageFlag +var _ context.Context + +// SearchMessagingFlags returns all matching MessagingFlags from store +func SearchMessagingFlags(ctx context.Context, s MessagingFlags, f types.MessageFlagFilter) (types.MessageFlagSet, types.MessageFlagFilter, error) { + return s.SearchMessagingFlags(ctx, f) +} + +// LookupMessagingFlagByID searches for flags by ID +func LookupMessagingFlagByID(ctx context.Context, s MessagingFlags, id uint64) (*types.MessageFlag, error) { + return s.LookupMessagingFlagByID(ctx, id) +} + +// LookupMessagingFlagByMessageIDUserIDFlag searches for flags by ID +func LookupMessagingFlagByMessageIDUserIDFlag(ctx context.Context, s MessagingFlags, message_id uint64, user_id uint64, flag string) (*types.MessageFlag, error) { + return s.LookupMessagingFlagByMessageIDUserIDFlag(ctx, message_id, user_id, flag) +} + +// CreateMessagingFlag creates one or more MessagingFlags in store +func CreateMessagingFlag(ctx context.Context, s MessagingFlags, rr ...*types.MessageFlag) error { + return s.CreateMessagingFlag(ctx, rr...) +} + +// UpdateMessagingFlag updates one or more (existing) MessagingFlags in store +func UpdateMessagingFlag(ctx context.Context, s MessagingFlags, rr ...*types.MessageFlag) error { + return s.UpdateMessagingFlag(ctx, rr...) +} + +// PartialMessagingFlagUpdate updates one or more existing MessagingFlags in store +func PartialMessagingFlagUpdate(ctx context.Context, s MessagingFlags, onlyColumns []string, rr ...*types.MessageFlag) error { + return s.PartialMessagingFlagUpdate(ctx, onlyColumns, rr...) +} + +// UpsertMessagingFlag creates new or updates existing one or more MessagingFlags in store +func UpsertMessagingFlag(ctx context.Context, s MessagingFlags, rr ...*types.MessageFlag) error { + return s.UpsertMessagingFlag(ctx, rr...) +} + +// DeleteMessagingFlag Deletes one or more MessagingFlags from store +func DeleteMessagingFlag(ctx context.Context, s MessagingFlags, rr ...*types.MessageFlag) error { + return s.DeleteMessagingFlag(ctx, rr...) +} + +// DeleteMessagingFlagByID Deletes MessagingFlag from store +func DeleteMessagingFlagByID(ctx context.Context, s MessagingFlags, ID uint64) error { + return s.DeleteMessagingFlagByID(ctx, ID) +} + +// TruncateMessagingFlags Deletes all MessagingFlags from store +func TruncateMessagingFlags(ctx context.Context, s MessagingFlags) error { + return s.TruncateMessagingFlags(ctx) +} diff --git a/store/messaging_flags.yaml b/store/messaging_flags.yaml new file mode 100644 index 000000000..2e4d03062 --- /dev/null +++ b/store/messaging_flags.yaml @@ -0,0 +1,33 @@ +import: + - github.com/cortezaproject/corteza-server/messaging/types + +types: + type: types.MessageFlag + +fields: + - { field: ID } + - { field: UserID } + - { field: MessageID } + - { field: ChannelID } + - { field: Flag } + - { field: CreatedAt } + - { field: DeletedAt } + +lookups: + - fields: [ ID ] + description: |- + searches for flags by ID + + - fields: [ MessageID, UserID, Flag ] + description: |- + searches for flags by ID + +rdbms: + alias: msg + table: messaging_message_flag + customFilterConverter: true + +search: + enablePaging: false + enableSorting: false + enableFilterCheckFunction: false diff --git a/store/messaging_mentions.gen.go b/store/messaging_mentions.gen.go new file mode 100644 index 000000000..027685271 --- /dev/null +++ b/store/messaging_mentions.gen.go @@ -0,0 +1,83 @@ +package store + +// This file is auto-generated. +// +// Template: pkg/codegen/assets/store_base.gen.go.tpl +// Definitions: store/messaging_mentions.yaml +// +// Changes to this file may cause incorrect behavior and will be lost if +// the code is regenerated. + +import ( + "context" + "github.com/cortezaproject/corteza-server/messaging/types" +) + +type ( + MessagingMentions interface { + SearchMessagingMentions(ctx context.Context, f types.MentionFilter) (types.MentionSet, types.MentionFilter, error) + LookupMessagingMentionByID(ctx context.Context, id uint64) (*types.Mention, error) + + CreateMessagingMention(ctx context.Context, rr ...*types.Mention) error + + UpdateMessagingMention(ctx context.Context, rr ...*types.Mention) error + PartialMessagingMentionUpdate(ctx context.Context, onlyColumns []string, rr ...*types.Mention) error + + UpsertMessagingMention(ctx context.Context, rr ...*types.Mention) error + + DeleteMessagingMention(ctx context.Context, rr ...*types.Mention) error + DeleteMessagingMentionByID(ctx context.Context, ID uint64) error + + TruncateMessagingMentions(ctx context.Context) error + } +) + +var _ *types.Mention +var _ context.Context + +// SearchMessagingMentions returns all matching MessagingMentions from store +func SearchMessagingMentions(ctx context.Context, s MessagingMentions, f types.MentionFilter) (types.MentionSet, types.MentionFilter, error) { + return s.SearchMessagingMentions(ctx, f) +} + +// LookupMessagingMentionByID searches for attachment by its ID +// +// It returns attachment even if deleted +func LookupMessagingMentionByID(ctx context.Context, s MessagingMentions, id uint64) (*types.Mention, error) { + return s.LookupMessagingMentionByID(ctx, id) +} + +// CreateMessagingMention creates one or more MessagingMentions in store +func CreateMessagingMention(ctx context.Context, s MessagingMentions, rr ...*types.Mention) error { + return s.CreateMessagingMention(ctx, rr...) +} + +// UpdateMessagingMention updates one or more (existing) MessagingMentions in store +func UpdateMessagingMention(ctx context.Context, s MessagingMentions, rr ...*types.Mention) error { + return s.UpdateMessagingMention(ctx, rr...) +} + +// PartialMessagingMentionUpdate updates one or more existing MessagingMentions in store +func PartialMessagingMentionUpdate(ctx context.Context, s MessagingMentions, onlyColumns []string, rr ...*types.Mention) error { + return s.PartialMessagingMentionUpdate(ctx, onlyColumns, rr...) +} + +// UpsertMessagingMention creates new or updates existing one or more MessagingMentions in store +func UpsertMessagingMention(ctx context.Context, s MessagingMentions, rr ...*types.Mention) error { + return s.UpsertMessagingMention(ctx, rr...) +} + +// DeleteMessagingMention Deletes one or more MessagingMentions from store +func DeleteMessagingMention(ctx context.Context, s MessagingMentions, rr ...*types.Mention) error { + return s.DeleteMessagingMention(ctx, rr...) +} + +// DeleteMessagingMentionByID Deletes MessagingMention from store +func DeleteMessagingMentionByID(ctx context.Context, s MessagingMentions, ID uint64) error { + return s.DeleteMessagingMentionByID(ctx, ID) +} + +// TruncateMessagingMentions Deletes all MessagingMentions from store +func TruncateMessagingMentions(ctx context.Context, s MessagingMentions) error { + return s.TruncateMessagingMentions(ctx) +} diff --git a/store/messaging_mentions.yaml b/store/messaging_mentions.yaml new file mode 100644 index 000000000..382553c24 --- /dev/null +++ b/store/messaging_mentions.yaml @@ -0,0 +1,30 @@ +import: + - github.com/cortezaproject/corteza-server/messaging/types + +types: + type: types.Mention + +fields: + - { field: ID } + - { field: MessageID } + - { field: ChannelID } + - { field: UserID } + - { field: MentionedByID } + - { field: CreatedAt } + +lookups: + - fields: [ ID ] + description: |- + searches for attachment by its ID + + It returns attachment even if deleted + +rdbms: + alias: msg + table: messaging_mention + customFilterConverter: true + +search: + enablePaging: false + enableSorting: false + enableFilterCheckFunction: false diff --git a/store/messaging_messages.gen.go b/store/messaging_messages.gen.go index 8942b8b69..1116694b2 100644 --- a/store/messaging_messages.gen.go +++ b/store/messaging_messages.gen.go @@ -29,6 +29,17 @@ type ( DeleteMessagingMessageByID(ctx context.Context, ID uint64) error TruncateMessagingMessages(ctx context.Context) error + + // Additional custom functions + + // SearchMessagingThreads (custom function) + SearchMessagingThreads(ctx context.Context, _filter types.MessageFilter) (types.MessageSet, types.MessageFilter, error) + + // CountMessagingMessagesFromID (custom function) + CountMessagingMessagesFromID(ctx context.Context, _channelID uint64, _threadID uint64, _lastReadMessageID uint64) (uint32, error) + + // LastMessagingMessageID (custom function) + LastMessagingMessageID(ctx context.Context, _channelID uint64, _threadID uint64) (uint64, error) } ) @@ -40,9 +51,9 @@ func SearchMessagingMessages(ctx context.Context, s MessagingMessages, f types.M return s.SearchMessagingMessages(ctx, f) } -// LookupMessagingMessageByID searches for attachment by its ID +// LookupMessagingMessageByID searches for message by its ID // -// It returns attachment even if deleted +// It returns message even if deleted func LookupMessagingMessageByID(ctx context.Context, s MessagingMessages, id uint64) (*types.Message, error) { return s.LookupMessagingMessageByID(ctx, id) } @@ -81,3 +92,15 @@ func DeleteMessagingMessageByID(ctx context.Context, s MessagingMessages, ID uin func TruncateMessagingMessages(ctx context.Context, s MessagingMessages) error { return s.TruncateMessagingMessages(ctx) } + +func SearchMessagingThreads(ctx context.Context, s MessagingMessages, _filter types.MessageFilter) (types.MessageSet, types.MessageFilter, error) { + return s.SearchMessagingThreads(ctx, _filter) +} + +func CountMessagingMessagesFromID(ctx context.Context, s MessagingMessages, _channelID uint64, _threadID uint64, _lastReadMessageID uint64) (uint32, error) { + return s.CountMessagingMessagesFromID(ctx, _channelID, _threadID, _lastReadMessageID) +} + +func LastMessagingMessageID(ctx context.Context, s MessagingMessages, _channelID uint64, _threadID uint64) (uint64, error) { + return s.LastMessagingMessageID(ctx, _channelID, _threadID) +} diff --git a/store/messaging_messages.yaml b/store/messaging_messages.yaml index 50a4a713f..32f8745ed 100644 --- a/store/messaging_messages.yaml +++ b/store/messaging_messages.yaml @@ -20,13 +20,32 @@ fields: lookups: - fields: [ ID ] description: |- - searches for attachment by its ID + searches for message by its ID - It returns attachment even if deleted + It returns message even if deleted + +functions: + - name: SearchMessagingThreads + arguments: + - { name: filter, + type: "types.MessageFilter" } + return: [ types.MessageSet, types.MessageFilter, error ] + - name: CountMessagingMessagesFromID + arguments: + - { name: channelID, type: uint64 } + - { name: threadID, type: uint64 } + - { name: lastReadMessageID, type: uint64 } + return: [ uint32, error ] + - name: LastMessagingMessageID + arguments: + - { name: channelID, type: uint64 } + - { name: threadID, type: uint64 } + return: [ uint64, error ] rdbms: alias: msg table: messaging_messages + customFilterConverter: true search: enablePaging: false diff --git a/store/messaging_unread.gen.go b/store/messaging_unread.gen.go index 120cb7c13..9f69bd49e 100644 --- a/store/messaging_unread.gen.go +++ b/store/messaging_unread.gen.go @@ -26,6 +26,23 @@ type ( DeleteMessagingUnreadByChannelIDReplyToUserID(ctx context.Context, channelID uint64, replyTo uint64, userID uint64) error TruncateMessagingUnreads(ctx context.Context) error + + // Additional custom functions + + // CountMessagingUnreadThreads (custom function) + CountMessagingUnreadThreads(ctx context.Context, _userID uint64, _channelID uint64) (types.UnreadSet, error) + + // CountMessagingUnread (custom function) + CountMessagingUnread(ctx context.Context, _userID uint64, _channelID uint64, _threadIDs ...uint64) (types.UnreadSet, error) + + // ResetMessagingUnreadThreads (custom function) + ResetMessagingUnreadThreads(ctx context.Context, _userID uint64, _channelID uint64) error + + // IncMessagingUnreadCount (custom function) + IncMessagingUnreadCount(ctx context.Context, _channelID uint64, _threadIDs uint64, _userID uint64) error + + // DecMessagingUnreadCount (custom function) + DecMessagingUnreadCount(ctx context.Context, _channelID uint64, _threadIDs uint64, _userID uint64) error } ) @@ -66,3 +83,23 @@ func DeleteMessagingUnreadByChannelIDReplyToUserID(ctx context.Context, s Messag func TruncateMessagingUnreads(ctx context.Context, s MessagingUnreads) error { return s.TruncateMessagingUnreads(ctx) } + +func CountMessagingUnreadThreads(ctx context.Context, s MessagingUnreads, _userID uint64, _channelID uint64) (types.UnreadSet, error) { + return s.CountMessagingUnreadThreads(ctx, _userID, _channelID) +} + +func CountMessagingUnread(ctx context.Context, s MessagingUnreads, _userID uint64, _channelID uint64, _threadIDs ...uint64) (types.UnreadSet, error) { + return s.CountMessagingUnread(ctx, _userID, _channelID, _threadIDs...) +} + +func ResetMessagingUnreadThreads(ctx context.Context, s MessagingUnreads, _userID uint64, _channelID uint64) error { + return s.ResetMessagingUnreadThreads(ctx, _userID, _channelID) +} + +func IncMessagingUnreadCount(ctx context.Context, s MessagingUnreads, _channelID uint64, _threadIDs uint64, _userID uint64) error { + return s.IncMessagingUnreadCount(ctx, _channelID, _threadIDs, _userID) +} + +func DecMessagingUnreadCount(ctx context.Context, s MessagingUnreads, _channelID uint64, _threadIDs uint64, _userID uint64) error { + return s.DecMessagingUnreadCount(ctx, _channelID, _threadIDs, _userID) +} diff --git a/store/messaging_unread.yaml b/store/messaging_unread.yaml index 9ddfedd97..4be017610 100644 --- a/store/messaging_unread.yaml +++ b/store/messaging_unread.yaml @@ -11,6 +11,40 @@ fields: - { field: LastMessageID, type: uint64 } - { field: Count, type: uint32 } + +functions: + - name: CountMessagingUnreadThreads + arguments: + - { name: userID, type: uint64 } + - { name: channelID, type: uint64 } + return: [ types.UnreadSet, error ] + + - name: CountMessagingUnread + arguments: + - { name: userID, type: uint64 } + - { name: channelID, type: uint64 } + - { name: threadIDs, type: ...uint64 } + return: [ types.UnreadSet, error ] + + - name: ResetMessagingUnreadThreads + arguments: + - { name: userID, type: uint64 } + - { name: channelID, type: uint64 } + return: [ error ] + + - name: IncMessagingUnreadCount + arguments: + - { name: channelID, type: uint64 } + - { name: threadIDs, type: uint64 } + - { name: userID, type: uint64 } + return: [ error ] + + - name: DecMessagingUnreadCount + arguments: + - { name: channelID, type: uint64 } + - { name: threadIDs, type: uint64 } + - { name: userID, type: uint64 } + return: [ error ] rdbms: alias: mur table: messaging_unread diff --git a/store/rdbms/messaging_flags.gen.go b/store/rdbms/messaging_flags.gen.go new file mode 100644 index 000000000..e8cd774a4 --- /dev/null +++ b/store/rdbms/messaging_flags.gen.go @@ -0,0 +1,326 @@ +package rdbms + +// This file is an auto-generated file +// +// Template: pkg/codegen/assets/store_rdbms.gen.go.tpl +// Definitions: store/messaging_flags.yaml +// +// Changes to this file may cause incorrect behavior +// and will be lost if the code is regenerated. + +import ( + "context" + "database/sql" + "errors" + "fmt" + "github.com/Masterminds/squirrel" + "github.com/cortezaproject/corteza-server/messaging/types" + "github.com/cortezaproject/corteza-server/store" +) + +var _ = errors.Is + +// SearchMessagingFlags returns all matching rows +// +// This function calls convertMessagingFlagFilter with the given +// types.MessageFlagFilter and expects to receive a working squirrel.SelectBuilder +func (s Store) SearchMessagingFlags(ctx context.Context, f types.MessageFlagFilter) (types.MessageFlagSet, types.MessageFlagFilter, error) { + var ( + err error + set []*types.MessageFlag + q squirrel.SelectBuilder + ) + q, err = s.convertMessagingFlagFilter(f) + if err != nil { + return nil, f, err + } + + return set, f, s.config.ErrorHandler(func() error { + set, _, _, err = s.QueryMessagingFlags(ctx, q, nil) + return err + + }()) +} + +// QueryMessagingFlags queries the database, converts and checks each row and +// returns collected set +// +// Fn also returns total number of fetched items and last fetched item so that the caller can construct cursor +// for next page of results +func (s Store) QueryMessagingFlags( + ctx context.Context, + q squirrel.Sqlizer, + check func(*types.MessageFlag) (bool, error), +) ([]*types.MessageFlag, uint, *types.MessageFlag, error) { + var ( + set = make([]*types.MessageFlag, 0, DefaultSliceCapacity) + res *types.MessageFlag + + // Query rows with + rows, err = s.Query(ctx, q) + + fetched uint + ) + + if err != nil { + return nil, 0, nil, err + } + + defer rows.Close() + for rows.Next() { + fetched++ + if err = rows.Err(); err == nil { + res, err = s.internalMessagingFlagRowScanner(rows) + } + + if err != nil { + return nil, 0, nil, err + } + + set = append(set, res) + } + + return set, fetched, res, rows.Err() +} + +// LookupMessagingFlagByID searches for flags by ID +func (s Store) LookupMessagingFlagByID(ctx context.Context, id uint64) (*types.MessageFlag, error) { + return s.execLookupMessagingFlag(ctx, squirrel.Eq{ + s.preprocessColumn("msg.id", ""): s.preprocessValue(id, ""), + }) +} + +// LookupMessagingFlagByMessageIDUserIDFlag searches for flags by ID +func (s Store) LookupMessagingFlagByMessageIDUserIDFlag(ctx context.Context, message_id uint64, user_id uint64, flag string) (*types.MessageFlag, error) { + return s.execLookupMessagingFlag(ctx, squirrel.Eq{ + s.preprocessColumn("msg.rel_message", ""): s.preprocessValue(message_id, ""), + s.preprocessColumn("msg.rel_user", ""): s.preprocessValue(user_id, ""), + s.preprocessColumn("msg.flag", ""): s.preprocessValue(flag, ""), + }) +} + +// CreateMessagingFlag creates one or more rows in messaging_message_flag table +func (s Store) CreateMessagingFlag(ctx context.Context, rr ...*types.MessageFlag) (err error) { + for _, res := range rr { + err = s.checkMessagingFlagConstraints(ctx, res) + if err != nil { + return err + } + + err = s.execCreateMessagingFlags(ctx, s.internalMessagingFlagEncoder(res)) + if err != nil { + return err + } + } + + return +} + +// UpdateMessagingFlag updates one or more existing rows in messaging_message_flag +func (s Store) UpdateMessagingFlag(ctx context.Context, rr ...*types.MessageFlag) error { + return s.config.ErrorHandler(s.PartialMessagingFlagUpdate(ctx, nil, rr...)) +} + +// PartialMessagingFlagUpdate updates one or more existing rows in messaging_message_flag +func (s Store) PartialMessagingFlagUpdate(ctx context.Context, onlyColumns []string, rr ...*types.MessageFlag) (err error) { + for _, res := range rr { + err = s.checkMessagingFlagConstraints(ctx, res) + if err != nil { + return err + } + + err = s.execUpdateMessagingFlags( + ctx, + squirrel.Eq{ + s.preprocessColumn("msg.id", ""): s.preprocessValue(res.ID, ""), + }, + s.internalMessagingFlagEncoder(res).Skip("id").Only(onlyColumns...)) + if err != nil { + return s.config.ErrorHandler(err) + } + } + + return +} + +// UpsertMessagingFlag updates one or more existing rows in messaging_message_flag +func (s Store) UpsertMessagingFlag(ctx context.Context, rr ...*types.MessageFlag) (err error) { + for _, res := range rr { + err = s.checkMessagingFlagConstraints(ctx, res) + if err != nil { + return err + } + + err = s.config.ErrorHandler(s.execUpsertMessagingFlags(ctx, s.internalMessagingFlagEncoder(res))) + if err != nil { + return err + } + } + + return nil +} + +// DeleteMessagingFlag Deletes one or more rows from messaging_message_flag table +func (s Store) DeleteMessagingFlag(ctx context.Context, rr ...*types.MessageFlag) (err error) { + for _, res := range rr { + + err = s.execDeleteMessagingFlags(ctx, squirrel.Eq{ + s.preprocessColumn("msg.id", ""): s.preprocessValue(res.ID, ""), + }) + if err != nil { + return s.config.ErrorHandler(err) + } + } + + return nil +} + +// DeleteMessagingFlagByID Deletes row from the messaging_message_flag table +func (s Store) DeleteMessagingFlagByID(ctx context.Context, ID uint64) error { + return s.execDeleteMessagingFlags(ctx, squirrel.Eq{ + s.preprocessColumn("msg.id", ""): s.preprocessValue(ID, ""), + }) +} + +// TruncateMessagingFlags Deletes all rows from the messaging_message_flag table +func (s Store) TruncateMessagingFlags(ctx context.Context) error { + return s.config.ErrorHandler(s.Truncate(ctx, s.messagingFlagTable())) +} + +// execLookupMessagingFlag prepares MessagingFlag query and executes it, +// returning types.MessageFlag (or error) +func (s Store) execLookupMessagingFlag(ctx context.Context, cnd squirrel.Sqlizer) (res *types.MessageFlag, err error) { + var ( + row rowScanner + ) + + row, err = s.QueryRow(ctx, s.messagingFlagsSelectBuilder().Where(cnd)) + if err != nil { + return + } + + res, err = s.internalMessagingFlagRowScanner(row) + if err != nil { + return + } + + return res, nil +} + +// execCreateMessagingFlags updates all matched (by cnd) rows in messaging_message_flag with given data +func (s Store) execCreateMessagingFlags(ctx context.Context, payload store.Payload) error { + return s.config.ErrorHandler(s.Exec(ctx, s.InsertBuilder(s.messagingFlagTable()).SetMap(payload))) +} + +// execUpdateMessagingFlags updates all matched (by cnd) rows in messaging_message_flag with given data +func (s Store) execUpdateMessagingFlags(ctx context.Context, cnd squirrel.Sqlizer, set store.Payload) error { + return s.config.ErrorHandler(s.Exec(ctx, s.UpdateBuilder(s.messagingFlagTable("msg")).Where(cnd).SetMap(set))) +} + +// execUpsertMessagingFlags inserts new or updates matching (by-primary-key) rows in messaging_message_flag with given data +func (s Store) execUpsertMessagingFlags(ctx context.Context, set store.Payload) error { + upsert, err := s.config.UpsertBuilder( + s.config, + s.messagingFlagTable(), + set, + "id", + ) + + if err != nil { + return err + } + + return s.config.ErrorHandler(s.Exec(ctx, upsert)) +} + +// execDeleteMessagingFlags Deletes all matched (by cnd) rows in messaging_message_flag with given data +func (s Store) execDeleteMessagingFlags(ctx context.Context, cnd squirrel.Sqlizer) error { + return s.config.ErrorHandler(s.Exec(ctx, s.DeleteBuilder(s.messagingFlagTable("msg")).Where(cnd))) +} + +func (s Store) internalMessagingFlagRowScanner(row rowScanner) (res *types.MessageFlag, err error) { + res = &types.MessageFlag{} + + if _, has := s.config.RowScanners["messagingFlag"]; has { + scanner := s.config.RowScanners["messagingFlag"].(func(_ rowScanner, _ *types.MessageFlag) error) + err = scanner(row, res) + } else { + err = row.Scan( + &res.ID, + &res.UserID, + &res.MessageID, + &res.ChannelID, + &res.Flag, + &res.CreatedAt, + &res.DeletedAt, + ) + } + + if err == sql.ErrNoRows { + return nil, store.ErrNotFound + } + + if err != nil { + return nil, fmt.Errorf("could not scan db row for MessagingFlag: %w", err) + } else { + return res, nil + } +} + +// QueryMessagingFlags returns squirrel.SelectBuilder with set table and all columns +func (s Store) messagingFlagsSelectBuilder() squirrel.SelectBuilder { + return s.SelectBuilder(s.messagingFlagTable("msg"), s.messagingFlagColumns("msg")...) +} + +// messagingFlagTable name of the db table +func (Store) messagingFlagTable(aa ...string) string { + var alias string + if len(aa) > 0 { + alias = " AS " + aa[0] + } + + return "messaging_message_flag" + alias +} + +// MessagingFlagColumns returns all defined table columns +// +// With optional string arg, all columns are returned aliased +func (Store) messagingFlagColumns(aa ...string) []string { + var alias string + if len(aa) > 0 { + alias = aa[0] + "." + } + + return []string{ + alias + "id", + alias + "rel_user", + alias + "rel_message", + alias + "rel_channel", + alias + "flag", + alias + "created_at", + alias + "deleted_at", + } +} + +// {true true false false false} + +// internalMessagingFlagEncoder encodes fields from types.MessageFlag to store.Payload (map) +// +// Encoding is done by using generic approach or by calling encodeMessagingFlag +// func when rdbms.customEncoder=true +func (s Store) internalMessagingFlagEncoder(res *types.MessageFlag) store.Payload { + return store.Payload{ + "id": res.ID, + "rel_user": res.UserID, + "rel_message": res.MessageID, + "rel_channel": res.ChannelID, + "flag": res.Flag, + "created_at": res.CreatedAt, + "deleted_at": res.DeletedAt, + } +} + +func (s *Store) checkMessagingFlagConstraints(ctx context.Context, res *types.MessageFlag) error { + + return nil +} diff --git a/store/rdbms/messaging_flags.go b/store/rdbms/messaging_flags.go new file mode 100644 index 000000000..2eb94c5c2 --- /dev/null +++ b/store/rdbms/messaging_flags.go @@ -0,0 +1,20 @@ +package rdbms + +import ( + "github.com/Masterminds/squirrel" + "github.com/cortezaproject/corteza-server/messaging/types" +) + +func (s Store) convertMessagingFlagFilter(f types.MessageFlagFilter) (query squirrel.SelectBuilder, err error) { + query = s.messagingFlagsSelectBuilder() + + if f.Flag != "" { + query.Where(squirrel.Eq{"flag": f.Flag}) + } + + if len(f.MessageID) > 0 { + query.Where(squirrel.Eq{"rel_message": f.MessageID}) + } + + return query, nil +} diff --git a/store/rdbms/messaging_mentions.gen.go b/store/rdbms/messaging_mentions.gen.go new file mode 100644 index 000000000..66d8d0746 --- /dev/null +++ b/store/rdbms/messaging_mentions.gen.go @@ -0,0 +1,316 @@ +package rdbms + +// This file is an auto-generated file +// +// Template: pkg/codegen/assets/store_rdbms.gen.go.tpl +// Definitions: store/messaging_mentions.yaml +// +// Changes to this file may cause incorrect behavior +// and will be lost if the code is regenerated. + +import ( + "context" + "database/sql" + "errors" + "fmt" + "github.com/Masterminds/squirrel" + "github.com/cortezaproject/corteza-server/messaging/types" + "github.com/cortezaproject/corteza-server/store" +) + +var _ = errors.Is + +// SearchMessagingMentions returns all matching rows +// +// This function calls convertMessagingMentionFilter with the given +// types.MentionFilter and expects to receive a working squirrel.SelectBuilder +func (s Store) SearchMessagingMentions(ctx context.Context, f types.MentionFilter) (types.MentionSet, types.MentionFilter, error) { + var ( + err error + set []*types.Mention + q squirrel.SelectBuilder + ) + q, err = s.convertMessagingMentionFilter(f) + if err != nil { + return nil, f, err + } + + return set, f, s.config.ErrorHandler(func() error { + set, _, _, err = s.QueryMessagingMentions(ctx, q, nil) + return err + + }()) +} + +// QueryMessagingMentions queries the database, converts and checks each row and +// returns collected set +// +// Fn also returns total number of fetched items and last fetched item so that the caller can construct cursor +// for next page of results +func (s Store) QueryMessagingMentions( + ctx context.Context, + q squirrel.Sqlizer, + check func(*types.Mention) (bool, error), +) ([]*types.Mention, uint, *types.Mention, error) { + var ( + set = make([]*types.Mention, 0, DefaultSliceCapacity) + res *types.Mention + + // Query rows with + rows, err = s.Query(ctx, q) + + fetched uint + ) + + if err != nil { + return nil, 0, nil, err + } + + defer rows.Close() + for rows.Next() { + fetched++ + if err = rows.Err(); err == nil { + res, err = s.internalMessagingMentionRowScanner(rows) + } + + if err != nil { + return nil, 0, nil, err + } + + set = append(set, res) + } + + return set, fetched, res, rows.Err() +} + +// LookupMessagingMentionByID searches for attachment by its ID +// +// It returns attachment even if deleted +func (s Store) LookupMessagingMentionByID(ctx context.Context, id uint64) (*types.Mention, error) { + return s.execLookupMessagingMention(ctx, squirrel.Eq{ + s.preprocessColumn("msg.id", ""): s.preprocessValue(id, ""), + }) +} + +// CreateMessagingMention creates one or more rows in messaging_mention table +func (s Store) CreateMessagingMention(ctx context.Context, rr ...*types.Mention) (err error) { + for _, res := range rr { + err = s.checkMessagingMentionConstraints(ctx, res) + if err != nil { + return err + } + + err = s.execCreateMessagingMentions(ctx, s.internalMessagingMentionEncoder(res)) + if err != nil { + return err + } + } + + return +} + +// UpdateMessagingMention updates one or more existing rows in messaging_mention +func (s Store) UpdateMessagingMention(ctx context.Context, rr ...*types.Mention) error { + return s.config.ErrorHandler(s.PartialMessagingMentionUpdate(ctx, nil, rr...)) +} + +// PartialMessagingMentionUpdate updates one or more existing rows in messaging_mention +func (s Store) PartialMessagingMentionUpdate(ctx context.Context, onlyColumns []string, rr ...*types.Mention) (err error) { + for _, res := range rr { + err = s.checkMessagingMentionConstraints(ctx, res) + if err != nil { + return err + } + + err = s.execUpdateMessagingMentions( + ctx, + squirrel.Eq{ + s.preprocessColumn("msg.id", ""): s.preprocessValue(res.ID, ""), + }, + s.internalMessagingMentionEncoder(res).Skip("id").Only(onlyColumns...)) + if err != nil { + return s.config.ErrorHandler(err) + } + } + + return +} + +// UpsertMessagingMention updates one or more existing rows in messaging_mention +func (s Store) UpsertMessagingMention(ctx context.Context, rr ...*types.Mention) (err error) { + for _, res := range rr { + err = s.checkMessagingMentionConstraints(ctx, res) + if err != nil { + return err + } + + err = s.config.ErrorHandler(s.execUpsertMessagingMentions(ctx, s.internalMessagingMentionEncoder(res))) + if err != nil { + return err + } + } + + return nil +} + +// DeleteMessagingMention Deletes one or more rows from messaging_mention table +func (s Store) DeleteMessagingMention(ctx context.Context, rr ...*types.Mention) (err error) { + for _, res := range rr { + + err = s.execDeleteMessagingMentions(ctx, squirrel.Eq{ + s.preprocessColumn("msg.id", ""): s.preprocessValue(res.ID, ""), + }) + if err != nil { + return s.config.ErrorHandler(err) + } + } + + return nil +} + +// DeleteMessagingMentionByID Deletes row from the messaging_mention table +func (s Store) DeleteMessagingMentionByID(ctx context.Context, ID uint64) error { + return s.execDeleteMessagingMentions(ctx, squirrel.Eq{ + s.preprocessColumn("msg.id", ""): s.preprocessValue(ID, ""), + }) +} + +// TruncateMessagingMentions Deletes all rows from the messaging_mention table +func (s Store) TruncateMessagingMentions(ctx context.Context) error { + return s.config.ErrorHandler(s.Truncate(ctx, s.messagingMentionTable())) +} + +// execLookupMessagingMention prepares MessagingMention query and executes it, +// returning types.Mention (or error) +func (s Store) execLookupMessagingMention(ctx context.Context, cnd squirrel.Sqlizer) (res *types.Mention, err error) { + var ( + row rowScanner + ) + + row, err = s.QueryRow(ctx, s.messagingMentionsSelectBuilder().Where(cnd)) + if err != nil { + return + } + + res, err = s.internalMessagingMentionRowScanner(row) + if err != nil { + return + } + + return res, nil +} + +// execCreateMessagingMentions updates all matched (by cnd) rows in messaging_mention with given data +func (s Store) execCreateMessagingMentions(ctx context.Context, payload store.Payload) error { + return s.config.ErrorHandler(s.Exec(ctx, s.InsertBuilder(s.messagingMentionTable()).SetMap(payload))) +} + +// execUpdateMessagingMentions updates all matched (by cnd) rows in messaging_mention with given data +func (s Store) execUpdateMessagingMentions(ctx context.Context, cnd squirrel.Sqlizer, set store.Payload) error { + return s.config.ErrorHandler(s.Exec(ctx, s.UpdateBuilder(s.messagingMentionTable("msg")).Where(cnd).SetMap(set))) +} + +// execUpsertMessagingMentions inserts new or updates matching (by-primary-key) rows in messaging_mention with given data +func (s Store) execUpsertMessagingMentions(ctx context.Context, set store.Payload) error { + upsert, err := s.config.UpsertBuilder( + s.config, + s.messagingMentionTable(), + set, + "id", + ) + + if err != nil { + return err + } + + return s.config.ErrorHandler(s.Exec(ctx, upsert)) +} + +// execDeleteMessagingMentions Deletes all matched (by cnd) rows in messaging_mention with given data +func (s Store) execDeleteMessagingMentions(ctx context.Context, cnd squirrel.Sqlizer) error { + return s.config.ErrorHandler(s.Exec(ctx, s.DeleteBuilder(s.messagingMentionTable("msg")).Where(cnd))) +} + +func (s Store) internalMessagingMentionRowScanner(row rowScanner) (res *types.Mention, err error) { + res = &types.Mention{} + + if _, has := s.config.RowScanners["messagingMention"]; has { + scanner := s.config.RowScanners["messagingMention"].(func(_ rowScanner, _ *types.Mention) error) + err = scanner(row, res) + } else { + err = row.Scan( + &res.ID, + &res.MessageID, + &res.ChannelID, + &res.UserID, + &res.MentionedByID, + &res.CreatedAt, + ) + } + + if err == sql.ErrNoRows { + return nil, store.ErrNotFound + } + + if err != nil { + return nil, fmt.Errorf("could not scan db row for MessagingMention: %w", err) + } else { + return res, nil + } +} + +// QueryMessagingMentions returns squirrel.SelectBuilder with set table and all columns +func (s Store) messagingMentionsSelectBuilder() squirrel.SelectBuilder { + return s.SelectBuilder(s.messagingMentionTable("msg"), s.messagingMentionColumns("msg")...) +} + +// messagingMentionTable name of the db table +func (Store) messagingMentionTable(aa ...string) string { + var alias string + if len(aa) > 0 { + alias = " AS " + aa[0] + } + + return "messaging_mention" + alias +} + +// MessagingMentionColumns returns all defined table columns +// +// With optional string arg, all columns are returned aliased +func (Store) messagingMentionColumns(aa ...string) []string { + var alias string + if len(aa) > 0 { + alias = aa[0] + "." + } + + return []string{ + alias + "id", + alias + "rel_message", + alias + "rel_channel", + alias + "rel_user", + alias + "rel_mentioned_by", + alias + "created_at", + } +} + +// {true true false false false} + +// internalMessagingMentionEncoder encodes fields from types.Mention to store.Payload (map) +// +// Encoding is done by using generic approach or by calling encodeMessagingMention +// func when rdbms.customEncoder=true +func (s Store) internalMessagingMentionEncoder(res *types.Mention) store.Payload { + return store.Payload{ + "id": res.ID, + "rel_message": res.MessageID, + "rel_channel": res.ChannelID, + "rel_user": res.UserID, + "rel_mentioned_by": res.MentionedByID, + "created_at": res.CreatedAt, + } +} + +func (s *Store) checkMessagingMentionConstraints(ctx context.Context, res *types.Mention) error { + + return nil +} diff --git a/store/rdbms/messaging_mentions.go b/store/rdbms/messaging_mentions.go new file mode 100644 index 000000000..ba031bb69 --- /dev/null +++ b/store/rdbms/messaging_mentions.go @@ -0,0 +1,16 @@ +package rdbms + +import ( + "github.com/Masterminds/squirrel" + "github.com/cortezaproject/corteza-server/messaging/types" +) + +func (s Store) convertMessagingMentionFilter(f types.MentionFilter) (query squirrel.SelectBuilder, err error) { + query = s.messagingMentionsSelectBuilder() + + if len(f.MessageID) > 0 { + query.Where(squirrel.Eq{"rel_message": f.MessageID}) + } + + return +} diff --git a/store/rdbms/messaging_messages.gen.go b/store/rdbms/messaging_messages.gen.go index 25d2b20e1..b98d258f7 100644 --- a/store/rdbms/messaging_messages.gen.go +++ b/store/rdbms/messaging_messages.gen.go @@ -30,7 +30,10 @@ func (s Store) SearchMessagingMessages(ctx context.Context, f types.MessageFilte set []*types.Message q squirrel.SelectBuilder ) - q = s.messagingMessagesSelectBuilder() + q, err = s.convertMessagingMessageFilter(f) + if err != nil { + return nil, f, err + } return set, f, s.config.ErrorHandler(func() error { set, _, _, err = s.QueryMessagingMessages(ctx, q, nil) @@ -80,9 +83,9 @@ func (s Store) QueryMessagingMessages( return set, fetched, res, rows.Err() } -// LookupMessagingMessageByID searches for attachment by its ID +// LookupMessagingMessageByID searches for message by its ID // -// It returns attachment even if deleted +// It returns message even if deleted func (s Store) LookupMessagingMessageByID(ctx context.Context, id uint64) (*types.Message, error) { return s.execLookupMessagingMessage(ctx, squirrel.Eq{ s.preprocessColumn("msg.id", ""): s.preprocessValue(id, ""), diff --git a/store/rdbms/messaging_messages.go b/store/rdbms/messaging_messages.go new file mode 100644 index 000000000..842394343 --- /dev/null +++ b/store/rdbms/messaging_messages.go @@ -0,0 +1,201 @@ +package rdbms + +import ( + "context" + "database/sql" + "github.com/Masterminds/squirrel" + "github.com/cortezaproject/corteza-server/messaging/types" + "strings" +) + +const ( + messagingMessagesMaxLimit = 100 +) + +func (s Store) convertMessagingMessageFilter(f types.MessageFilter) (query squirrel.SelectBuilder, err error) { + if f.Limit == 0 || f.Limit > messagingMessagesMaxLimit { + f.Limit = messagingMessagesMaxLimit + } + + query = s.messagingMessagesSelectBuilder() + + if f.Query != "" { + q := "%" + strings.ToLower(f.Query) + "%" + query = query.Where(squirrel.Like{"LOWER(m.message)": q}) + } + + if len(f.ChannelID) > 0 { + query = query.Where(squirrel.Eq{"m.rel_channel": f.ChannelID}) + } + + if len(f.UserID) > 0 { + query = query.Where(squirrel.Eq{"m.rel_user": f.UserID}) + } + + if len(f.ThreadID) > 0 { + query = query.Where(squirrel.Eq{"m.reply_to": f.ThreadID}) + } else { + query = query.Where(squirrel.Eq{"m.reply_to": 0}) + } + + if f.AttachmentsOnly { + // Override Type filter + f.Type = []string{ + types.MessageTypeAttachment.String(), + types.MessageTypeInlineImage.String(), + } + } + + if len(f.Type) > 0 { + query = query.Where(squirrel.Eq{"m.type": f.Type}) + } + + // first, exclusive + if f.AfterID > 0 { + query = query.OrderBy("m.id ASC") + query = query.Where(squirrel.Gt{"m.id": f.AfterID}) + } + + // from, inclusive + if f.FromID > 0 { + query = query.OrderBy("m.id ASC") + query = query.Where(squirrel.GtOrEq{"m.id": f.FromID}) + } + + // last, exclusive + if f.BeforeID > 0 { + query = query.OrderBy("m.id DESC") + query = query.Where(squirrel.Lt{"m.id": f.BeforeID}) + } + + // to, inclusive + if f.ToID > 0 { + query = query.OrderBy("m.id DESC") + query = query.Where(squirrel.LtOrEq{"m.id": f.ToID}) + } + + if f.BookmarkedOnly || f.PinnedOnly { + var ( + flagQuery squirrel.SelectBuilder + flag = types.MessageFlagBookmarkedMessage + ) + if f.PinnedOnly { + flag = types.MessageFlagPinnedToChannel + } + + if flagQuery, err = s.convertMessagingFlagFilter(types.MessageFlagFilter{Flag: flag}); err != nil { + return + } else { + query = query. + Where(squirrel.Eq{"m.id": flagQuery}) + } + } + + query = query. + OrderBy("id DESC"). + Limit(uint64(f.Limit)) + return +} + +func (s Store) SearchMessagingThreads(ctx context.Context, filter types.MessageFilter) (set types.MessageSet, f types.MessageFilter, err error) { + if f.Limit == 0 || f.Limit > messagingMessagesMaxLimit { + f.Limit = messagingMessagesMaxLimit + } + + // Selecting first valid (deleted_at IS NULL) messages in threads (replies > 0 && reply_to = 0) + // that belong to filtered channels and we've contributed to (or stated it) + originals := squirrel. + Select("id AS original_id"). + From(s.messagingMessageTable()). + Where(squirrel.And{ + squirrel.Eq{ + "deleted_at": nil, + "rel_channel": f.ChannelID, + "reply_to": 0, + }, + squirrel.Gt{"replies": 0}, + squirrel.Or{ + squirrel.Eq{"rel_user": filter.CurrentUserID}, + squirrel.Expr( + "id IN (SELECT DISTINCT reply_to FROM messaging_message WHERE rel_user = ?)", + filter.CurrentUserID), + }, + }). + OrderBy("id DESC"). + Limit(uint64(f.Limit)) + + // Prepare the actual message selector + base := s.messagingMessagesSelectBuilder(). + Where(squirrel.Eq{"m.deleted_at": nil}). + Join("originals ON (original_id IN (id, reply_to))") + + if f.Query != "" { + q := "%" + strings.ToLower(f.Query) + "%" + base = base.Where(squirrel.Like{"LOWER(m.message)": q}) + } + + // Create CTE with originals & base + cte := squirrel.ConcatExpr("WITH originals AS (", originals, ") ", base) + + if set, _, _, err = s.QueryMessagingMessages(ctx, cte, nil); err != nil { + return nil, f, err + } + + return set, f, nil +} + +// CountMessagingMessagesFromID returns number of messages from last message on +func (s Store) CountMessagingMessagesFromID(ctx context.Context, channelID, threadID, messageID uint64) (c uint32, err error) { + if messageID == 0 { + // No need for counting, zero unread messages... + return + } + + var ( + row *sql.Row + //rval := struct{ Count uint32 }{} + q = s.SelectBuilder(s.messagingMessageTable("msg"), "COUNT(*)"). + Where(squirrel.NotEq{"type": types.MessageTypeChannelEvent}). + Where(squirrel.Gt{"id": messageID}). + Where(squirrel.Eq{ + "rel_channel": channelID, + "reply_to": threadID, + "deleted_at": nil, + }) + ) + + if row, err = s.QueryRow(ctx, q); err != nil { + return + } + + if err = row.Scan(&c); err != nil { + return + } + + return +} + +// LastMessagingMessageID returns last message in the channel or thread +func (s Store) LastMessagingMessageID(ctx context.Context, channelID, threadID uint64) (ID uint64, err error) { + var ( + row *sql.Row + //rval := struct{ Count uint32 }{} + q = s.SelectBuilder(s.messagingMessageTable("msg"), "COALESCE(MAX(id), 0)"). + Where(squirrel.NotEq{"type": types.MessageTypeChannelEvent}). + Where(squirrel.Eq{ + "rel_channel": channelID, + "reply_to": threadID, + "deleted_at": nil, + }) + ) + + if row, err = s.QueryRow(ctx, q); err != nil { + return + } + + if err = row.Scan(&ID); err != nil { + return + } + + return +} diff --git a/store/rdbms/messaging_unread.go b/store/rdbms/messaging_unread.go new file mode 100644 index 000000000..c465c4476 --- /dev/null +++ b/store/rdbms/messaging_unread.go @@ -0,0 +1,139 @@ +package rdbms + +import ( + "context" + "database/sql" + "github.com/Masterminds/squirrel" + "github.com/cortezaproject/corteza-server/messaging/types" + "github.com/cortezaproject/corteza-server/store" +) + +// CountReplies counts unread thread info +func (s Store) CountMessagingUnreadThreads(ctx context.Context, userID, channelID uint64) (uu types.UnreadSet, err error) { + var ( + rows *sql.Rows + + q = squirrel.Select( + "rel_channel", + "rel_user", + "sum(count) AS count", + "sum(CASE WHEN count > 0 THEN 1 ELSE 0 END) AS total", + ). + From(s.messagingUnreadTable()). + Where("rel_reply_to > 0 AND count > 0"). + GroupBy("rel_channel", "rel_user") + ) + + if userID > 0 { + q = q.Where(squirrel.Eq{"rel_user": userID}) + } + + if channelID > 0 { + q = q.Where(squirrel.Eq{"rel_channel": channelID}) + } + + if rows, err = s.Query(ctx, q); err != nil { + return nil, err + } + + defer rows.Close() + + uu = make([]*types.Unread, 0, 512) + for rows.Next() { + u := &types.Unread{} + if err = rows.Scan(&u.ChannelID, &u.UserID, &u.ThreadCount, &u.ThreadTotal); err != nil { + return + } + + uu = append(uu, u) + } + + return uu, nil +} + +// Count returns counts unread channel info +func (s Store) CountMessagingUnread(ctx context.Context, userID, channelID uint64, threadIDs ...uint64) (uu types.UnreadSet, err error) { + var ( + q = squirrel. + Select( + "rel_channel", + "rel_last_message", + "rel_user", + "rel_reply_to", + "count", + ). + From(s.messagingUnreadTable()) + ) + + if userID > 0 { + q = q.Where(squirrel.Eq{"rel_user": userID}) + } + + if channelID > 0 { + q = q.Where(squirrel.Eq{"rel_channel": channelID}) + } + + if len(threadIDs) == 0 { + q = q.Where(squirrel.Eq{"rel_reply_to": 0}) + } else { + q = q.Where(squirrel.Eq{"rel_reply_to": threadIDs}) + } + + uu, _, _, err = s.QueryMessagingUnreads(ctx, q, nil) + return +} + +func (s Store) ResetMessagingUnreadThreads(ctx context.Context, userID, channelID uint64) error { + var ( + cnd = squirrel.And{ + squirrel.Eq{"rel_user": userID, "rel_channel": channelID}, + squirrel.Gt{"rel_reply_to": 0}, + } + set = store.Payload{"count": 0} + ) + + return s.execUpdateComposeRecordValues(ctx, cnd, set) + +} + +func (s Store) IncMessagingUnreadCount(ctx context.Context, channelID uint64, threadID uint64, userID uint64) error { + var ( + cnd = squirrel.And{ + squirrel.Eq{"rel_reply_to": threadID, "rel_channel": channelID}, + squirrel.NotEq{"rel_user": userID}, + } + + upd = s.UpdateBuilder(s.messagingUnreadTable()).Where(cnd).Set("count", squirrel.Expr("count + 1")) + ) + + return s.config.ErrorHandler(s.Exec(ctx, upd)) + +} + +func (s Store) DecMessagingUnreadCount(ctx context.Context, channelID uint64, threadID uint64, userID uint64) (err error) { + var ( + cnd = squirrel.And{ + squirrel.Eq{"rel_reply_to": threadID, "rel_channel": channelID}, + squirrel.Gt{"count": 0}, + } + + upd = s.UpdateBuilder(s.messagingUnreadTable()).Where(cnd).Set("count", squirrel.Expr("count - 1")) + ) + + if err = s.Exec(ctx, upd); err != nil { + return s.config.ErrorHandler(err) + } + + err = s.UpsertMessagingUnread(ctx, &types.Unread{ + ChannelID: channelID, + ReplyTo: threadID, + UserID: userID, + Count: 0, + }) + + if err != nil { + return s.config.ErrorHandler(err) + } + + return nil +} diff --git a/store/tests/gen_test.go b/store/tests/gen_test.go index 50dde2c40..6787ad047 100644 --- a/store/tests/gen_test.go +++ b/store/tests/gen_test.go @@ -17,6 +17,8 @@ package tests // - store/messaging_attachments.yaml // - store/messaging_channel_members.yaml // - store/messaging_channels.yaml +// - store/messaging_flags.yaml +// - store/messaging_mentions.yaml // - store/messaging_messages.yaml // - store/messaging_unread.yaml // - store/rbac_rules.yaml @@ -112,6 +114,16 @@ func testAllGenerated(t *testing.T, s store.Storable) { testMessagingChannels(t, s) }) + // Run generated tests for MessagingFlags + t.Run("MessagingFlags", func(t *testing.T) { + testMessagingFlags(t, s) + }) + + // Run generated tests for MessagingMentions + t.Run("MessagingMentions", func(t *testing.T) { + testMessagingMentions(t, s) + }) + // Run generated tests for MessagingMessages t.Run("MessagingMessages", func(t *testing.T) { testMessagingMessages(t, s)