DeforaNetworks/corteza
3
0
Fork 0
Code Activity

Migrated messaging message to store

Browse Source
This commit is contained in:
Denis Arh
2020-09-01 21:31:25 +02:00
parent 8dc5315b75
commit 3540b4f128
29 changed files with 1949 additions and 543 deletions
Download Patch File Download Diff File Expand all files Collapse all files
compose/service/attachment.go
+2
View File
@@ -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,
}
}
messaging/repository/attachment.go
+133 -133
View File
@@ -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)
//}
messaging/repository/error.go
-3
View File
@@ -9,9 +9,6 @@ type (
)
const (
ErrDatabaseError = repositoryError("DatabaseError")
ErrNotImplemented = repositoryError("NotImplemented")
ErrConfigError = repositoryError("ConfigError")
ErrEventsPullClosed = repositoryError("EventsPullClosed")
)
messaging/repository/mention.go
+87 -87
View File
@@ -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})
//}
messaging/repository/unread.go
+13 -13
View File
@@ -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
messaging/rest/attachment.go
+6 -7
View File
@@ -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)
messaging/service/attachment.go
+39 -44
View File
@@ -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
messaging/service/channel.go
+2 -2
View File
@@ -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()
messaging/service/message.go
+249 -225
View File
@@ -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()
messaging/service/service.go
+1 -1
View File
@@ -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
}
messaging/types/mention.go
+15 -14
View File
@@ -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
}
)
messaging/types/message_flag.go
+12 -7
View File
@@ -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
}
)
store/interfaces.gen.go
+4
View File
@@ -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
store/messaging_flags.gen.go
+87
View File
@@ -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)
}
store/messaging_flags.yaml
+33
View File
@@ -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
store/messaging_mentions.gen.go
+83
View File
@@ -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)
}
store/messaging_mentions.yaml
+30
View File
@@ -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
store/messaging_messages.gen.go
+25 -2
View File
@@ -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)
}
store/messaging_messages.yaml
+21 -2
View File
@@ -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
store/messaging_unread.gen.go
+37
View File
@@ -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)
}
store/messaging_unread.yaml
+34
View File
@@ -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
store/rdbms/messaging_flags.gen.go
+326
View File
@@ -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
}
store/rdbms/messaging_flags.go
+20
View File
@@ -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
}
store/rdbms/messaging_mentions.gen.go
+316
View File
@@ -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
}
store/rdbms/messaging_mentions.go
+16
View File
@@ -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
}
store/rdbms/messaging_messages.gen.go
+6 -3
View File
@@ -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, ""),
store/rdbms/messaging_messages.go
+201
View File
@@ -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
}
store/rdbms/messaging_unread.go
+139
View File
@@ -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
}
store/tests/gen_test.go
+12
View File
@@ -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)