Merge pull request #7 from volmedo/track-last-notified-message

Track last notified message

cmd/almendruco/main.go

@@ -1,7 +1,6 @@
 package main
 
 import (
-	"errors"
 	"fmt"
 	"log"
 	"strconv"
@@ -66,13 +65,19 @@ func notifyMessages(r repo.Repo, rc raices.Client, n notifier.Notifier) error {
 			return fmt.Errorf("error fetching messages from Raíces: %s", err)
 		}
 
-		if len(msgs) == 0 {
-			//lint:ignore ST1005 the word "Raíces" is capitalized as it is the name of the application
-			return errors.New("Raíces client returned no messages")
-		}
-
-		if err := n.Notify(notifier.ChatID(chatID), msgs); err != nil {
-			return fmt.Errorf("error notifying messages: %s", err)
+		if len(msgs) != 0 {
+			last, err := n.Notify(notifier.ChatID(chatID), msgs)
+			if err != nil {
+				// Notify notifies messages until it encounters an error, so even in the case of an error
+				// happening we can still update last notified message to avoid notifying again messages
+				// that have already been notified
+				_ = r.UpdateLastNotifiedMessage(strconv.FormatUint(chatID, 10), last)
+				return fmt.Errorf("error notifying messages: %s", err)
+			}
+
+			if err := r.UpdateLastNotifiedMessage(strconv.FormatUint(chatID, 10), last); err != nil {
+				return fmt.Errorf("error updating last notified message: %s", err)
+			}
 		}
 	}
 

internal/notifier/notifier.go

@@ -5,5 +5,5 @@ import "github.com/volmedo/almendruco.git/internal/raices"
 type ChatID uint64
 
 type Notifier interface {
-	Notify(chatID ChatID, msgs []raices.Message) error
+	Notify(chatID ChatID, msgs []raices.Message) (uint64, error)
 }

internal/notifier/telegram.go

@@ -41,31 +41,34 @@ func NewTelegramNotifier(baseURL, botToken string) (Notifier, error) {
 	}, nil
 }
 
-func (tn *telegramNotifier) Notify(chatID ChatID, msgs []raices.Message) error {
+func (tn *telegramNotifier) Notify(chatID ChatID, msgs []raices.Message) (uint64, error) {
 	u, _ := url.Parse(tn.baseURL.String())
 	u.Path = path.Join(u.Path, sendMessagePath)
 
 	params := url.Values{}
 	params.Set(chatIDParam, strconv.FormatUint(uint64(chatID), 10))
 	params.Set(parseModeParam, parseModeHTML)
 
+	var lastNotifiedMessage uint64
 	for _, m := range msgs {
 		text := formatText(m)
 
 		params.Set(textParam, text)
 
 		resp, err := tn.http.Post(u.String(), "application/x-www-form-urlencoded", strings.NewReader(params.Encode()))
 		if err != nil {
-			return err
+			return lastNotifiedMessage, err
 		}
 		defer resp.Body.Close()
 
 		if resp.StatusCode != http.StatusOK {
-			return fmt.Errorf("received status code %d", resp.StatusCode)
+			return lastNotifiedMessage, fmt.Errorf("received status code %d", resp.StatusCode)
 		}
+
+		lastNotifiedMessage = m.ID
 	}
 
-	return nil
+	return lastNotifiedMessage, nil
 }
 
 func formatText(m raices.Message) string {

internal/notifier/telegram_test.go

@@ -39,7 +39,7 @@ func TestNotify(t *testing.T) {
 		assert.Equal(t, "123456789", reqChatID)
 
 		reqText := r.Form.Get(textParam)
-		expectedText := "New message arrived to Raíces!\nFrom: Test Sender\nSubject: Test Subject\nHi you, this is a test message\nAttachments: true"
+		expectedText := "Nuevo mensaje en Raíces!\n\n<b>Fecha:</b> 11/11/2021 00:00\n<b>De:</b> Test Sender\n<b>Asunto:</b> Test Subject\n\nHi you, this is a test message\n\n<b>Adjuntos:</b>\n\t\t\tattachment.file\n"
 		assert.Equal(t, expectedText, reqText)
 
 		w.WriteHeader(http.StatusOK)
@@ -49,6 +49,7 @@ func TestNotify(t *testing.T) {
 	tn, err := NewTelegramNotifier(svr.URL, "test_token")
 	require.NoError(t, err)
 
-	err = tn.Notify(chatID, []raices.Message{msg})
+	lastNotifiedMessage, err := tn.Notify(chatID, []raices.Message{msg})
 	assert.NoError(t, err)
+	assert.Equal(t, uint64(123456), lastNotifiedMessage)
 }

internal/raices/client.go

@@ -83,12 +83,12 @@ func (c *client) FetchMessages(creds repo.Credentials, lastNotifiedMessage uint6
 			return []Message{}, err
 		}
 
-		msgs = append(msgs, parsed...)
-
 		numMsgs = len(parsed)
+
+		msgs = append(msgs, parsed...)
 	}
 
-	return msgs, nil
+	return reverse(msgs), nil
 }
 
 func (c *client) login(creds repo.Credentials) error {
@@ -178,3 +178,12 @@ func parse(raw []rawMessage) ([]Message, error) {
 
 	return parsed, nil
 }
+
+func reverse(msgs []Message) []Message {
+	reversed := make([]Message, 0, len(msgs))
+	for i := len(msgs) - 1; i >= 0; i-- {
+		reversed = append(reversed, msgs[i])
+	}
+
+	return reversed
+}

internal/raices/models.go

@@ -29,6 +29,17 @@ type rawMessage struct {
 	ReadDate            string       `json:"F_LECTURA"`
 }
 
+type Message struct {
+	ID                  uint64
+	SentDate            time.Time
+	Sender              string
+	Subject             string
+	Body                string
+	ContainsAttachments bool
+	Attachments         []Attachment
+	ReadDate            time.Time
+}
+
 type Attachment struct {
 	ID       uint64 `json:"X_ADJMENSAL"`
 	FileName string `json:"T_NOMFIC"`
@@ -70,14 +81,3 @@ func parseMessage(rm rawMessage) (Message, error) {
 		ReadDate:            readDate,
 	}, nil
 }
-
-type Message struct {
-	ID                  uint64
-	SentDate            time.Time
-	Sender              string
-	Subject             string
-	Body                string
-	ContainsAttachments bool
-	Attachments         []Attachment
-	ReadDate            time.Time
-}

internal/repo/dynamodbrepo/dynamodb.go

@@ -1,8 +1,8 @@
 package dynamodbrepo
 
 import (
-	"errors"
 	"fmt"
+	"strconv"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/session"
@@ -54,5 +54,25 @@ func (dr *dynamoDBRepo) GetChats() ([]repo.Chat, error) {
 }
 
 func (dr *dynamoDBRepo) UpdateLastNotifiedMessage(chatID string, lastNotifiedMessage uint64) error {
-	return errors.New("not implemented yet")
+	input := &dynamodb.UpdateItemInput{
+		ExpressionAttributeValues: map[string]*dynamodb.AttributeValue{
+			":last": {
+				N: aws.String(strconv.FormatUint(lastNotifiedMessage, 10)),
+			},
+		},
+		Key: map[string]*dynamodb.AttributeValue{
+			"id": {
+				S: aws.String(chatID),
+			},
+		},
+		TableName:        aws.String(tableName),
+		UpdateExpression: aws.String("SET lastNotifiedMessage = :last"),
+	}
+
+	_, err := dr.db.UpdateItem(input)
+	if err != nil {
+		return fmt.Errorf("update last notified message failed: %s", err)
+	}
+
+	return nil
 }

internal/repo/dynamodbrepo/dynamodb_test.go

@@ -1,6 +1,8 @@
 package dynamodbrepo
 
 import (
+	"fmt"
+	"strconv"
 	"testing"
 
 	"github.com/aws/aws-sdk-go/service/dynamodb"
@@ -48,6 +50,33 @@ func (m *dynamoDBClientMock) Scan(input *dynamodb.ScanInput) (*dynamodb.ScanOutp
 	}, nil
 }
 
+func (m *dynamoDBClientMock) UpdateItem(input *dynamodb.UpdateItemInput) (*dynamodb.UpdateItemOutput, error) {
+	lastStr := input.ExpressionAttributeValues[":last"].N
+	last, err := strconv.ParseUint(*lastStr, 10, 64)
+	if err != nil {
+		return nil, fmt.Errorf("bad lastNotifiedMessage value: %s", *lastStr)
+	}
+
+	expectedLast := uint64(11)
+	if last != expectedLast {
+		return nil, fmt.Errorf("expected last to be %d, but got %d", expectedLast, last)
+	}
+
+	key := input.Key["id"].S
+	expectedKey := "some_chat"
+	if *key != expectedKey {
+		return nil, fmt.Errorf("expected key to be %s but got %s", expectedKey, *key)
+	}
+
+	updateExp := input.UpdateExpression
+	expectedExp := "SET lastNotifiedMessage = :last"
+	if *updateExp != expectedExp {
+		return nil, fmt.Errorf("expected update exp to be \"%s\" but got \"%s\"", expectedExp, *updateExp)
+	}
+
+	return &dynamodb.UpdateItemOutput{}, nil
+}
+
 func TestGetChats(t *testing.T) {
 	mockClient := &dynamoDBClientMock{}
 	dynamoRepo := NewRepoWithClient(mockClient)
@@ -66,5 +95,5 @@ func TestUpdateLastNotifiedMessage(t *testing.T) {
 
 	err := dynamoRepo.UpdateLastNotifiedMessage("some_chat", 11)
 
-	assert.Error(t, err)
+	assert.NoError(t, err)
 }