heimoshuiyu/postmoogle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

BREAKING: update mautrix to 0.15.x

Browse Source
This commit is contained in:
Aine
2023-06-01 14:32:20 +00:00
parent a6b20a75ab
commit 2bdb8ca635
222 changed files with 7851 additions and 23986 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
bot/access.go
View File

@@ -105,11 +105,11 @@ func (b *Bot) IsGreylisted(addr net.Addr) bool {
greylist := b.cfg.GetGreylist()
greylistedAt, ok := greylist.Get(addr)
if !ok {
b.log.Debug("greylisting %s", addr.String())
b.log.Debug().Str("addr", addr.String()).Msg("greylisting")
greylist.Add(addr)
err := b.cfg.SetGreylist(greylist)
if err != nil {
b.log.Error("cannot update greylist with %s: %v", addr.String(), err)
b.log.Error().Err(err).Str("addr", addr.String()).Msg("cannot update greylist")
}
return true
}
@@ -128,7 +128,7 @@ func (b *Bot) IsTrusted(addr net.Addr) bool {
ip := utils.AddrIP(addr)
for _, proxy := range b.proxies {
if ip == proxy {
b.log.Debug("address %s is trusted", ip)
b.log.Debug().Str("addr", ip).Msg("address is trusted")
return true
}
}
@@ -144,12 +144,12 @@ func (b *Bot) Ban(addr net.Addr) {
if b.IsTrusted(addr) {
return
}
b.log.Debug("attempting to ban %s", addr.String())
b.log.Debug().Str("addr", addr.String()).Msg("attempting to ban")
banlist := b.cfg.GetBanlist()
banlist.Add(addr)
err := b.cfg.SetBanlist(banlist)
if err != nil {
b.log.Error("cannot update banlist with %s: %v", addr.String(), err)
b.log.Error().Err(err).Str("addr", addr.String()).Msg("cannot update banlist")
}
}
@@ -172,18 +172,18 @@ func (b *Bot) AllowAuth(email, password string) (id.RoomID, bool) {
}
cfg, err := b.cfg.GetRoom(roomID)
if err != nil {
b.log.Error("failed to retrieve settings: %v", err)
b.log.Error().Err(err).Msg("failed to retrieve settings")
return "", false
}
if cfg.NoSend() {
b.log.Warn("trying to send email from %q (%q), but it's receive-only", email, roomID)
b.log.Warn().Str("email", email).Str("roomID", roomID.String()).Msg("trying to send email, but room is receive-only")
return "", false
}
allow, err := argon2pw.CompareHashWithPassword(cfg.Password(), password)
if err != nil {
b.log.Warn("Password for %s is not valid: %v", email, err)
b.log.Warn().Err(err).Str("email", email).Msg("Password is not valid")
}
return roomID, allow
}

6
bot/activation.go
View File

@@ -27,14 +27,14 @@ func (b *Bot) ActivateMailbox(ownerID id.UserID, roomID id.RoomID, mailbox strin
}
func (b *Bot) activateNone(ownerID id.UserID, roomID id.RoomID, mailbox string) bool {
b.log.Debug("activating mailbox %q (%q) of %q through flow 'none'", mailbox, roomID, ownerID)
b.log.Debug().Str("mailbox", mailbox).Str("roomID", roomID.String()).Str("ownerID", ownerID.String()).Msg("activating mailbox through the flow 'none'")
b.rooms.Store(mailbox, roomID)
return true
}
func (b *Bot) activateNotify(ownerID id.UserID, roomID id.RoomID, mailbox string) bool {
b.log.Debug("activating mailbox %q (%q) of %q through flow 'notify'", mailbox, roomID, ownerID)
b.log.Debug().Str("mailbox", mailbox).Str("roomID", roomID.String()).Str("ownerID", ownerID.String()).Msg("activating mailbox through the flow 'notify'")
b.rooms.Store(mailbox, roomID)
if len(b.adminRooms) == 0 {
return true
@@ -45,7 +45,7 @@ func (b *Bot) activateNotify(ownerID id.UserID, roomID id.RoomID, mailbox string
content := format.RenderMarkdown(msg, true, true)
_, err := b.lp.Send(adminRoom, &content)
if err != nil {
b.log.Info("cannot send mailbox activation notification to the admin room %q", adminRoom)
b.log.Info().Str("adminRoom", adminRoom.String()).Msg("cannot send mailbox activation notification to the admin room")
continue
}
break

19
bot/bot.go
View File

@@ -7,7 +7,7 @@ import (
"sync"
"github.com/getsentry/sentry-go"
"gitlab.com/etke.cc/go/logger"
"github.com/rs/zerolog"
"gitlab.com/etke.cc/linkpearl"
"maunium.net/go/mautrix/event"
"maunium.net/go/mautrix/format"
@@ -32,12 +32,13 @@ type Bot struct {
allowedUsers []*regexp.Regexp
allowedAdmins []*regexp.Regexp
adminRooms []id.RoomID
ignoreBefore int64 // mautrix 0.15.x migration
commands commandList
rooms sync.Map
proxies []string
sendmail func(string, string, string) error
cfg *config.Manager
log *logger.Logger
log *zerolog.Logger
lp *linkpearl.Linkpearl
mu utils.Mutex
q *queue.Queue
@@ -48,7 +49,7 @@ type Bot struct {
func New(
q *queue.Queue,
lp *linkpearl.Linkpearl,
log *logger.Logger,
log *zerolog.Logger,
cfg *config.Manager,
proxies []string,
prefix string,
@@ -92,12 +93,9 @@ func New(
// Error message to the log and matrix room
func (b *Bot) Error(ctx context.Context, roomID id.RoomID, message string, args ...interface{}) {
b.log.Error(message, args...)
err := fmt.Errorf(message, args...)
b.log.Error().Err(err).Msg("something is wrong")
if hub := sentry.GetHubFromContext(ctx); hub != nil {
sentry.GetHubFromContext(ctx).CaptureException(err)
}
if roomID != "" {
b.SendError(ctx, roomID, err.Error())
}
@@ -120,15 +118,16 @@ func (b *Bot) SendNotice(ctx context.Context, roomID id.RoomID, message string)
// Start performs matrix /sync
func (b *Bot) Start(statusMsg string) error {
if err := b.migrate(); err != nil {
if err := b.migrateMautrix015(); err != nil {
return err
}
if err := b.syncRooms(); err != nil {
return err
}
b.initSync()
b.log.Info("Postmoogle has been started")
b.log.Info().Msg("Postmoogle has been started")
return b.lp.Start(statusMsg)
}
@@ -136,7 +135,7 @@ func (b *Bot) Start(statusMsg string) error {
func (b *Bot) Stop() {
err := b.lp.GetClient().SetPresence(event.PresenceOffline)
if err != nil {
b.log.Error("cannot set presence = offline: %v", err)
b.log.Error().Err(err).Msg("cannot set presence = offline")
}
b.lp.GetClient().StopSync()
}

8
bot/command.go
View File

@@ -295,7 +295,7 @@ func (b *Bot) handle(ctx context.Context) {
evt := eventFromContext(ctx)
err := b.lp.GetClient().MarkRead(evt.RoomID, evt.ID)
if err != nil {
b.log.Error("cannot send read receipt: %v", err)
b.log.Error().Err(err).Msg("cannot send read receipt")
}
content := evt.Content.AsMessage()
@@ -322,7 +322,7 @@ func (b *Bot) handle(ctx context.Context) {
}
_, err = b.lp.GetClient().UserTyping(evt.RoomID, true, 30*time.Second)
if err != nil {
b.log.Error("cannot send typing notification: %v", err)
b.log.Error().Err(err).Msg("cannot send typing notification")
}
defer b.lp.GetClient().UserTyping(evt.RoomID, false, 30*time.Second) //nolint:errcheck
@@ -406,7 +406,7 @@ func (b *Bot) sendHelp(ctx context.Context) {
cfg, serr := b.cfg.GetRoom(evt.RoomID)
if serr != nil {
b.log.Error("cannot retrieve settings: %v", serr)
b.log.Error().Err(serr).Msg("cannot retrieve settings")
}
var msg strings.Builder
@@ -517,7 +517,7 @@ func (b *Bot) runSend(ctx context.Context) {
}
queued, err := b.Sendmail(evt.ID, from, to, data)
if queued {
b.log.Error("cannot send email: %v", err)
b.log.Error().Err(err).Msg("cannot send email")
b.saveSentMetadata(ctx, queued, evt.ID, recipients, eml, cfg)
continue
}

2
bot/command_admin.go
View File

@@ -37,7 +37,7 @@ func (b *Bot) sendMailboxes(ctx context.Context) {
}
config, err := b.cfg.GetRoom(roomID)
if err != nil {
b.log.Error("cannot retrieve settings: %v", err)
b.log.Error().Err(err).Msg("cannot retrieve settings")
}
mailboxes[mailbox] = config

24
bot/config/bot.go
View File

@@ -13,15 +13,16 @@ const acBotKey = "cc.etke.postmoogle.config"
// bot options keys
const (
BotAdminRoom = "adminroom"
BotUsers = "users"
BotCatchAll = "catch-all"
BotDKIMSignature = "dkim.pub"
BotDKIMPrivateKey = "dkim.pem"
BotQueueBatch = "queue:batch"
BotQueueRetries = "queue:retries"
BotBanlistEnabled = "banlist:enabled"
BotGreylist = "greylist"
BotAdminRoom = "adminroom"
BotUsers = "users"
BotCatchAll = "catch-all"
BotDKIMSignature = "dkim.pub"
BotDKIMPrivateKey = "dkim.pem"
BotQueueBatch = "queue:batch"
BotQueueRetries = "queue:retries"
BotBanlistEnabled = "banlist:enabled"
BotGreylist = "greylist"
BotMautrix015Migration = "mautrix015migration"
)
// Bot map
@@ -37,6 +38,11 @@ func (s Bot) Set(key, value string) {
s[strings.ToLower(strings.TrimSpace(key))] = value
}
// Mautrix015Migration option (timestamp)
func (s Bot) Mautrix015Migration() int64 {
return utils.Int64(s.Get(BotMautrix015Migration))
}
// Users option
func (s Bot) Users() []string {
value := s.Get(BotUsers)

12
bot/config/manager.go
View File

@@ -1,7 +1,7 @@
package config
import (
"gitlab.com/etke.cc/go/logger"
"github.com/rs/zerolog"
"gitlab.com/etke.cc/linkpearl"
"maunium.net/go/mautrix/id"
@@ -11,12 +11,12 @@ import (
// Manager of configs
type Manager struct {
mu utils.Mutex
log *logger.Logger
log *zerolog.Logger
lp *linkpearl.Linkpearl
}
// New config manager
func New(lp *linkpearl.Linkpearl, log *logger.Logger) *Manager {
func New(lp *linkpearl.Linkpearl, log *zerolog.Logger) *Manager {
m := &Manager{
mu: utils.NewMutex(),
lp: lp,
@@ -32,7 +32,7 @@ func (m *Manager) GetBot() Bot {
var config Bot
config, err = m.lp.GetAccountData(acBotKey)
if err != nil {
m.log.Error("cannot get bot settings: %v", utils.UnwrapError(err))
m.log.Error().Err(utils.UnwrapError(err)).Msg("cannot get bot settings")
}
if config == nil {
config = make(Bot, 0)
@@ -72,7 +72,7 @@ func (m *Manager) GetBanlist() List {
defer m.mu.Unlock("banlist")
config, err := m.lp.GetAccountData(acBanlistKey)
if err != nil {
m.log.Error("cannot get banlist: %v", utils.UnwrapError(err))
m.log.Error().Err(utils.UnwrapError(err)).Msg("cannot get banlist")
}
if config == nil {
config = make(List, 0)
@@ -100,7 +100,7 @@ func (m *Manager) SetBanlist(cfg List) error {
func (m *Manager) GetGreylist() List {
config, err := m.lp.GetAccountData(acGreylistKey)
if err != nil {
m.log.Error("cannot get banlist: %v", utils.UnwrapError(err))
m.log.Error().Err(utils.UnwrapError(err)).Msg("cannot get banlist")
}
if config == nil {
config = make(List, 0)

61
bot/data.go
View File

@@ -1,42 +1,14 @@
package bot
import (
"strconv"
"time"
"maunium.net/go/mautrix/id"
"gitlab.com/etke.cc/postmoogle/bot/config"
)
var migrations = []string{}
func (b *Bot) migrate() error {
b.log.Debug("migrating database...")
tx, beginErr := b.lp.GetDB().Begin()
if beginErr != nil {
b.log.Error("cannot begin transaction: %v", beginErr)
return beginErr
}
for _, query := range migrations {
_, execErr := tx.Exec(query)
if execErr != nil {
b.log.Error("cannot apply migration: %v", execErr)
// nolint // we already have the execErr to return
tx.Rollback()
return execErr
}
}
commitErr := tx.Commit()
if commitErr != nil {
b.log.Error("cannot commit transaction: %v", commitErr)
// nolint // we already have the commitErr to return
tx.Rollback()
return commitErr
}
return nil
}
func (b *Bot) syncRooms() error {
adminRooms := []id.RoomID{}
@@ -73,7 +45,7 @@ func (b *Bot) syncRooms() error {
func (b *Bot) migrateRoomSettings(roomID id.RoomID) {
cfg, err := b.cfg.GetRoom(roomID)
if err != nil {
b.log.Error("cannot retrieve room settings: %v", err)
b.log.Error().Err(err).Msg("cannot retrieve room settings")
return
}
if _, ok := cfg[config.RoomActive]; !ok {
@@ -86,10 +58,33 @@ func (b *Bot) migrateRoomSettings(roomID id.RoomID) {
cfg.MigrateSpamlistSettings()
err = b.cfg.SetRoom(roomID, cfg)
if err != nil {
b.log.Error("cannot migrate room settings: %v", err)
b.log.Error().Err(err).Msg("cannot migrate room settings")
}
}
// migrateMautrix015 adds a special timestamp to bot's config
// to ignore any message events happened before that timestamp
// with migration to maturix 0.15.x the state store has been changed
// alongside with other database configs to simplify maintenance,
// but with that simplification there is no proper way to migrate
// existing sync token and session info. No data loss, tho.
func (b *Bot) migrateMautrix015() error {
cfg := b.cfg.GetBot()
ts := cfg.Mautrix015Migration()
// already migrated
if ts > 0 {
b.ignoreBefore = ts
return nil
}
ts = time.Now().UTC().UnixMilli()
b.ignoreBefore = ts
tss := strconv.FormatInt(ts, 10)
cfg.Set(config.BotMautrix015Migration, tss)
return b.cfg.SetBot(cfg)
}
func (b *Bot) initBotUsers() ([]string, error) {
cfg := b.cfg.GetBot()
cfgUsers := cfg.Users()

33
bot/email.go
View File

@@ -44,7 +44,7 @@ func (b *Bot) Sendmail(eventID id.EventID, from, to, data string) (bool, error)
err := b.sendmail(from, to, data)
if err != nil {
if strings.HasPrefix(err.Error(), "4") {
b.log.Info("email %s (from=%s to=%s) was added to the queue: %v", eventID, from, to, err)
b.log.Info().Err(err).Str("id", eventID.String()).Str("from", from).Str("to", to).Msg("email has been added to the queue")
return true, b.q.Add(eventID.String(), from, to, data)
}
return false, err
@@ -90,7 +90,7 @@ func (b *Bot) GetMapping(mailbox string) (id.RoomID, bool) {
func (b *Bot) GetIFOptions(roomID id.RoomID) email.IncomingFilteringOptions {
cfg, err := b.cfg.GetRoom(roomID)
if err != nil {
b.log.Error("cannot retrieve room settings: %v", err)
b.log.Error().Err(err).Msg("cannot retrieve room settings")
}
return cfg
@@ -185,7 +185,7 @@ func (b *Bot) SendEmailReply(ctx context.Context) {
meta.MessageID = email.MessageID(evt.ID, meta.FromDomain)
meta.References = meta.References + " " + meta.MessageID
b.log.Info("sending email reply: %+v", meta)
b.log.Info().Any("meta", meta).Msg("sending email reply")
eml := email.New(meta.MessageID, meta.InReplyTo, meta.References, meta.Subject, meta.From, meta.To, meta.RcptTo, meta.CC, body, htmlBody, nil, nil)
data := eml.Compose(b.cfg.GetBot().DKIMPrivateKey())
if data == "" {
@@ -198,7 +198,7 @@ func (b *Bot) SendEmailReply(ctx context.Context) {
for _, to := range recipients {
queued, err = b.Sendmail(evt.ID, meta.From, to, data)
if queued {
b.log.Error("cannot send email: %v", err)
b.log.Error().Err(err).Msg("cannot send email")
b.saveSentMetadata(ctx, queued, meta.ThreadID, recipients, eml, cfg)
continue
}
@@ -300,36 +300,33 @@ func (e *parentEmail) calculateRecipients(from string) {
func (b *Bot) getParentEvent(evt *event.Event) (id.EventID, *event.Event) {
content := evt.Content.AsMessage()
threadID := utils.EventParent(evt.ID, content)
b.log.Debug("looking up for the parent event of %s within thread %s", evt.ID, threadID)
b.log.Debug().Str("eventID", evt.ID.String()).Str("threadID", threadID.String()).Msg("looking up for the parent event within thread")
if threadID == evt.ID {
b.log.Debug("event %s is the thread itself")
b.log.Debug().Str("eventID", evt.ID.String()).Msg("event is the thread itself")
return threadID, evt
}
lastEventID := b.getLastEventID(evt.RoomID, threadID)
b.log.Debug("the last event of the thread %s (and parent of the %s) is %s", threadID, evt.ID, lastEventID)
b.log.Debug().Str("eventID", evt.ID.String()).Str("threadID", threadID.String()).Str("lastEventID", lastEventID.String()).Msg("the last event of the thread (and parent of the event) has been found")
if lastEventID == evt.ID {
return threadID, evt
}
parentEvt, err := b.lp.GetClient().GetEvent(evt.RoomID, lastEventID)
if err != nil {
b.log.Error("cannot get parent event: %v", err)
b.log.Error().Err(err).Msg("cannot get parent event")
return threadID, nil
}
utils.ParseContent(parentEvt, parentEvt.Type)
b.log.Debug("type of the parsed content is: %T", parentEvt.Content.Parsed)
if !b.lp.GetStore().IsEncrypted(evt.RoomID) {
b.log.Debug("found the last event (plaintext) of the thread %s (and parent of the %s): %+v", threadID, evt.ID, parentEvt)
if !b.lp.GetMachine().StateStore.IsEncrypted(evt.RoomID) {
return threadID, parentEvt
}
decrypted, err := b.lp.GetMachine().DecryptMegolmEvent(parentEvt)
decrypted, err := b.lp.GetClient().Crypto.Decrypt(parentEvt)
if err != nil {
b.log.Error("cannot decrypt parent event: %v", err)
b.log.Error().Err(err).Msg("cannot decrypt parent event")
return threadID, nil
}
b.log.Debug("found the last event (decrypted) of the thread %s (and parent of the %s): %+v", threadID, evt.ID, parentEvt)
return threadID, decrypted
}
@@ -422,7 +419,7 @@ func (b *Bot) getThreadID(roomID id.RoomID, messageID string, references string)
key := acMessagePrefix + "." + refID
data, err := b.lp.GetRoomAccountData(roomID, key)
if err != nil {
b.log.Error("cannot retrieve thread ID from %s: %v", key, err)
b.log.Error().Err(err).Str("key", key).Msg("cannot retrieve thread ID")
continue
}
if data["eventID"] != "" {
@@ -437,7 +434,7 @@ func (b *Bot) setThreadID(roomID id.RoomID, messageID string, eventID id.EventID
key := acMessagePrefix + "." + messageID
err := b.lp.SetRoomAccountData(roomID, key, map[string]string{"eventID": eventID.String()})
if err != nil {
b.log.Error("cannot save thread ID to %s: %v", key, err)
b.log.Error().Err(err).Str("key", key).Msg("cannot save thread ID")
}
}
@@ -445,7 +442,7 @@ func (b *Bot) getLastEventID(roomID id.RoomID, threadID id.EventID) id.EventID {
key := acLastEventPrefix + "." + threadID.String()
data, err := b.lp.GetRoomAccountData(roomID, key)
if err != nil {
b.log.Error("cannot retrieve last event ID from %s: %v", key, err)
b.log.Error().Err(err).Str("key", key).Msg("cannot retrieve last event ID")
return threadID
}
if data["eventID"] != "" {
@@ -459,6 +456,6 @@ func (b *Bot) setLastEventID(roomID id.RoomID, threadID id.EventID, eventID id.E
key := acLastEventPrefix + "." + threadID.String()
err := b.lp.SetRoomAccountData(roomID, key, map[string]string{"eventID": eventID.String()})
if err != nil {
b.log.Error("cannot save thread ID to %s: %v", key, err)
b.log.Error().Err(err).Str("key", key).Msg("cannot save thread ID")
}
}

22
bot/queue/manager.go
View File

@@ -1,7 +1,7 @@
package queue
import (
"gitlab.com/etke.cc/go/logger"
"github.com/rs/zerolog"
"gitlab.com/etke.cc/linkpearl"
"gitlab.com/etke.cc/postmoogle/bot/config"
@@ -10,8 +10,8 @@ import (
const (
acQueueKey = "cc.etke.postmoogle.mailqueue"
defaultQueueBatch = 1
defaultQueueRetries = 3
defaultQueueBatch = 10
defaultQueueRetries = 100
)
// Queue manager
@@ -19,12 +19,12 @@ type Queue struct {
mu utils.Mutex
lp *linkpearl.Linkpearl
cfg *config.Manager
log *logger.Logger
log *zerolog.Logger
sendmail func(string, string, string) error
}
// New queue
func New(lp *linkpearl.Linkpearl, cfg *config.Manager, log *logger.Logger) *Queue {
func New(lp *linkpearl.Linkpearl, cfg *config.Manager, log *zerolog.Logger) *Queue {
return &Queue{
mu: utils.Mutex{},
lp: lp,
@@ -40,7 +40,7 @@ func (q *Queue) SetSendmail(function func(string, string, string) error) {
// Process queue
func (q *Queue) Process() {
q.log.Debug("staring queue processing...")
q.log.Debug().Msg("staring queue processing...")
cfg := q.cfg.GetBot()
batchSize := cfg.QueueBatch()
@@ -57,23 +57,23 @@ func (q *Queue) Process() {
defer q.mu.Unlock(acQueueKey)
index, err := q.lp.GetAccountData(acQueueKey)
if err != nil {
q.log.Error("cannot get queue index: %v", err)
q.log.Error().Err(err).Msg("cannot get queue index")
}
i := 0
for id, itemkey := range index {
if i > batchSize {
q.log.Debug("finished re-deliveries from queue")
q.log.Debug().Msg("finished re-deliveries from queue")
return
}
if dequeue := q.try(itemkey, maxRetries); dequeue {
q.log.Info("email %q has been delivered", id)
q.log.Info().Str("id", id).Msg("email has been delivered")
err = q.Remove(id)
if err != nil {
q.log.Error("cannot dequeue email %q: %v", id, err)
q.log.Error().Err(err).Str("id", id).Msg("cannot dequeue email")
}
}
i++
}
q.log.Debug("ended queue processing")
q.log.Debug().Msg("ended queue processing")
}

22
bot/queue/queue.go
View File

@@ -19,7 +19,7 @@ func (q *Queue) Add(id, from, to, data string) error {
defer q.mu.Unlock(itemkey)
err := q.lp.SetAccountData(itemkey, item)
if err != nil {
q.log.Error("cannot enqueue email id=%q: %v", id, err)
q.log.Error().Err(err).Str("id", id).Msg("cannot enqueue email")
return err
}
@@ -27,13 +27,13 @@ func (q *Queue) Add(id, from, to, data string) error {
defer q.mu.Unlock(acQueueKey)
queueIndex, err := q.lp.GetAccountData(acQueueKey)
if err != nil {
q.log.Error("cannot get queue index: %v", err)
q.log.Error().Err(err).Msg("cannot get queue index")
return err
}
queueIndex[id] = itemkey
err = q.lp.SetAccountData(acQueueKey, queueIndex)
if err != nil {
q.log.Error("cannot save queue index: %v", err)
q.log.Error().Err(err).Msg("cannot save queue index")
return err
}
@@ -44,7 +44,7 @@ func (q *Queue) Add(id, from, to, data string) error {
func (q *Queue) Remove(id string) error {
index, err := q.lp.GetAccountData(acQueueKey)
if err != nil {
q.log.Error("cannot get queue index: %v", err)
q.log.Error().Err(err).Msg("cannot get queue index")
return err
}
itemkey := index[id]
@@ -54,7 +54,7 @@ func (q *Queue) Remove(id string) error {
delete(index, id)
err = q.lp.SetAccountData(acQueueKey, index)
if err != nil {
q.log.Error("cannot update queue index: %v", err)
q.log.Error().Err(err).Msg("cannot update queue index")
return err
}
@@ -70,13 +70,13 @@ func (q *Queue) try(itemkey string, maxRetries int) bool {
item, err := q.lp.GetAccountData(itemkey)
if err != nil {
q.log.Error("cannot retrieve a queue item %q: %v", itemkey, err)
q.log.Error().Err(err).Str("id", itemkey).Msg("cannot retrieve a queue item")
return false
}
q.log.Debug("processing queue item %+v", item)
q.log.Debug().Any("item", item).Msg("processing queue item")
attempts, err := strconv.Atoi(item["attempts"])
if err != nil {
q.log.Error("cannot parse attempts of %q: %v", itemkey, err)
q.log.Error().Err(err).Str("id", itemkey).Msg("cannot parse attempts")
return false
}
if attempts > maxRetries {
@@ -85,16 +85,16 @@ func (q *Queue) try(itemkey string, maxRetries int) bool {
err = q.sendmail(item["from"], item["to"], item["data"])
if err == nil {
q.log.Info("email %q from queue was delivered")
q.log.Info().Str("id", itemkey).Msg("email from queue was delivered")
return true
}
q.log.Info("attempted to deliver email id=%q, retry=%q, but it's not ready yet: %v", item["id"], item["attempts"], err)
q.log.Info().Str("id", itemkey).Str("from", item["from"]).Str("to", item["to"]).Err(err).Msg("attempted to deliver email, but it's not ready yet")
attempts++
item["attempts"] = strconv.Itoa(attempts)
err = q.lp.SetAccountData(itemkey, item)
if err != nil {
q.log.Error("cannot update attempt count on email %q: %v", itemkey, err)
q.log.Error().Err(err).Str("id", itemkey).Msg("cannot update attempt count on email")
}
return false

51
bot/sync.go
View File

@@ -22,17 +22,12 @@ func (b *Bot) initSync() {
func(_ mautrix.EventSource, evt *event.Event) {
go b.onMessage(evt)
})
b.lp.OnEventType(
event.EventEncrypted,
func(_ mautrix.EventSource, evt *event.Event) {
go b.onEncryptedMessage(evt)
})
}
// joinPermit is called by linkpearl when processing "invite" events and deciding if rooms should be auto-joined or not
func (b *Bot) joinPermit(evt *event.Event) bool {
if !mxidwc.Match(evt.Sender.String(), b.allowedUsers) {
b.log.Debug("Rejecting room invitation from unallowed user: %s", evt.Sender)
b.log.Debug().Str("userID", evt.Sender.String()).Msg("Rejecting room invitation from unallowed user")
return false
}
@@ -40,6 +35,11 @@ func (b *Bot) joinPermit(evt *event.Event) bool {
}
func (b *Bot) onMembership(evt *event.Event) {
// mautrix 0.15.x migration
if b.ignoreBefore >= evt.Timestamp {
return
}
ctx := newContext(evt)
evtType := evt.Content.AsMember().Membership
@@ -60,31 +60,15 @@ func (b *Bot) onMessage(evt *event.Event) {
if evt.Sender == b.lp.GetClient().UserID {
return
}
// mautrix 0.15.x migration
if b.ignoreBefore >= evt.Timestamp {
return
}
ctx := newContext(evt)
b.handle(ctx)
}
func (b *Bot) onEncryptedMessage(evt *event.Event) {
// ignore own messages
if evt.Sender == b.lp.GetClient().UserID {
return
}
// ignore encrypted events in noecryption mode
if b.lp.GetMachine() == nil {
return
}
ctx := newContext(evt)
decrypted, err := b.lp.GetMachine().DecryptMegolmEvent(evt)
if err != nil {
b.Error(ctx, evt.RoomID, "cannot decrypt a message: %v", err)
return
}
ctx = eventToContext(ctx, decrypted)
b.handle(ctx)
}
// onBotJoin handles the "bot joined the room" event
func (b *Bot) onBotJoin(ctx context.Context) {
evt := eventFromContext(ctx)
@@ -92,7 +76,7 @@ func (b *Bot) onBotJoin(ctx context.Context) {
// as described in this bug report: https://github.com/matrix-org/synapse/issues/9768
_, ok := b.handledMembershipEvents.LoadOrStore(evt.ID, true)
if ok {
b.log.Info("Suppressing already handled event %s", evt.ID)
b.log.Info().Str("eventID", evt.ID.String()).Msg("Suppressing already handled event")
return
}
@@ -104,13 +88,18 @@ func (b *Bot) onLeave(ctx context.Context) {
evt := eventFromContext(ctx)
_, ok := b.handledMembershipEvents.LoadOrStore(evt.ID, true)
if ok {
b.log.Info("Suppressing already handled event %s", evt.ID)
b.log.Info().Str("eventID", evt.ID.String()).Msg("Suppressing already handled event")
return
}
members := b.lp.GetStore().GetRoomMembers(evt.RoomID)
members, err := b.lp.GetClient().StateStore.GetRoomJoinedOrInvitedMembers(evt.RoomID)
if err != nil {
b.log.Error().Err(err).Str("roomID", evt.RoomID.String()).Msg("cannot get joined or invited members")
return
}
count := len(members)
if count == 1 && members[0] == b.lp.GetClient().UserID {
b.log.Info("no more users left in the %s room", evt.RoomID)
b.log.Info().Str("roomID", evt.RoomID.String()).Msg("no more users left in the room")
b.runStop(ctx)
_, err := b.lp.GetClient().LeaveRoom(evt.RoomID)
if err != nil {