module: added mam events

CHANGELOG.md

@@ -4,8 +4,8 @@
 
 ## 0.62.0 (2022/09/13)
 
-* [FEATURE] module: added support for xep-0313 [#241](https://github.com/ortuman/jackal/pull/241)
-* [ENHANCEMENT] auth: re-enable TLS 1.3 channel binding during auth using [#247](https://github.com/ortuman/jackal/pull/247)
+* [FEATURE] module: added support for xep-0313 [#241](https://github.com/ortuman/jackal/pull/241), [#253](https://github.com/ortuman/jackal/pull/253)
+* [ENHANCEMENT] auth: re-enable TLS 1.3 channel binding during auth using [RFC 9266](https://www.rfc-editor.org/rfc/rfc9266) [#247](https://github.com/ortuman/jackal/pull/247)
 * [ENHANCEMENT] hook: include propagated context into execution parameter. [#249](https://github.com/ortuman/jackal/pull/249)
 * [ENHANCEMENT] transport: limit writer buffer size [#251](https://github.com/ortuman/jackal/pull/251)
 

pkg/hook/mam.go

@@ -0,0 +1,39 @@
+// Copyright 2022 The jackal Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package hook
+
+import (
+	archivemodel "github.com/ortuman/jackal/pkg/model/archive"
+)
+
+const (
+	// ArchiveMessageQueried hook runs whenever an archive is queried.
+	ArchiveMessageQueried = "mam.message.queried"
+
+	// ArchiveMessageArchived hook runs whenever a message is archived.
+	ArchiveMessageArchived = "mam.message.archieved"
+)
+
+// MamInfo contains all information associated to a mam (XEP-0313) event.
+type MamInfo struct {
+	// ArchiveID is the id of the mam archive associated to this event.
+	ArchiveID string
+
+	// Message is the message stanza associated to this event.
+	Message *archivemodel.Message
+
+	// Filters contains filters applied to the archive queried event.
+	Filters *archivemodel.Filters
+}

pkg/module/xep0313/mam.go

@@ -284,6 +284,13 @@ func (m *Mam) sendArchiveMessages(ctx context.Context, iq *stravaganza.IQ) error
 		_, _ = m.router.Route(ctx, xmpputil.MakeErrorStanza(iq, stanzaerror.InternalServerError))
 		return err
 	}
+	// run archive queried event
+	if err := m.runHook(ctx, hook.ArchiveMessageQueried, &hook.MamInfo{
+		ArchiveID: archiveID,
+		Filters:   filters,
+	}); err != nil {
+		return err
+	}
 
 	// return not found error if any requested id cannot be found
 	switch {
@@ -433,20 +440,28 @@ func (m *Mam) handleRoutedMessage(execCtx *hook.ExecutionContext, elem stravagan
 }
 
 func (m *Mam) archiveMessage(ctx context.Context, message *stravaganza.Message, archiveID, id string) error {
-	return m.rep.InTransaction(ctx, func(ctx context.Context, tx repository.Transaction) error {
-		err := tx.InsertArchiveMessage(ctx, &archivemodel.Message{
-			ArchiveId: archiveID,
-			Id:        id,
-			FromJid:   message.FromJID().String(),
-			ToJid:     message.ToJID().String(),
-			Message:   message.Proto(),
-			Stamp:     timestamppb.Now(),
-		})
+	archiveMsg := &archivemodel.Message{
+		ArchiveId: archiveID,
+		Id:        id,
+		FromJid:   message.FromJID().String(),
+		ToJid:     message.ToJID().String(),
+		Message:   message.Proto(),
+		Stamp:     timestamppb.Now(),
+	}
+	err := m.rep.InTransaction(ctx, func(ctx context.Context, tx repository.Transaction) error {
+		err := tx.InsertArchiveMessage(ctx, archiveMsg)
 		if err != nil {
 			return err
 		}
 		return tx.DeleteArchiveOldestMessages(ctx, archiveID, m.cfg.QueueSize)
 	})
+	if err != nil {
+		return err
+	}
+	return m.runHook(ctx, hook.ArchiveMessageArchived, &hook.MamInfo{
+		ArchiveID: archiveID,
+		Message:   archiveMsg,
+	})
 }
 
 func (m *Mam) addRecipientStanzaID(originalMsg *stravaganza.Message) *stravaganza.Message {
@@ -458,6 +473,15 @@ func (m *Mam) addRecipientStanzaID(originalMsg *stravaganza.Message) *stravaganz
 	return xmpputil.MakeStanzaIDMessage(originalMsg, archiveID, toJID.ToBareJID().String())
 }
 
+func (m *Mam) runHook(ctx context.Context, hookName string, inf *hook.MamInfo) error {
+	_, err := m.hk.Run(hookName, &hook.ExecutionContext{
+		Info:    inf,
+		Sender:  m,
+		Context: ctx,
+	})
+	return err
+}
+
 // IsArchiveRequested determines whether archive has been requested over a C2S stream by inspecting inf parameter.
 func IsArchiveRequested(inf c2smodel.Info) bool {
 	return inf.Bool(archiveRequestedCtxKey)

pkg/module/xep0313/mam_test.go

@@ -101,6 +101,7 @@ func TestMam_Metadata(t *testing.T) {
 	}
 	mam := &Mam{
 		rep:    repMock,
+		hk:     hook.NewHooks(),
 		router: routerMock,
 		logger: kitlog.NewNopLogger(),
 	}
@@ -274,6 +275,7 @@ func TestMam_SendArchiveMessages(t *testing.T) {
 
 	mam := &Mam{
 		rep:    repMock,
+		hk:     hook.NewHooks(),
 		router: routerMock,
 		logger: kitlog.NewNopLogger(),
 	}