ntfy/message/store_test.go

package message_test

import (
	"net/netip"
	"sort"
	"sync"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"heckel.io/ntfy/v2/message"
	"heckel.io/ntfy/v2/model"
)

func testCacheMessages(t *testing.T, s message.Store) {
	m1 := model.NewDefaultMessage("mytopic", "my message")
	m1.Time = 1

	m2 := model.NewDefaultMessage("mytopic", "my other message")
	m2.Time = 2

	require.Nil(t, s.AddMessage(m1))
	require.Nil(t, s.AddMessage(model.NewDefaultMessage("example", "my example message")))
	require.Nil(t, s.AddMessage(m2))

	// Adding invalid
	require.Equal(t, model.ErrUnexpectedMessageType, s.AddMessage(model.NewKeepaliveMessage("mytopic"))) // These should not be added!
	require.Equal(t, model.ErrUnexpectedMessageType, s.AddMessage(model.NewOpenMessage("example")))      // These should not be added!

	// mytopic: count
	counts, err := s.MessageCounts()
	require.Nil(t, err)
	require.Equal(t, 2, counts["mytopic"])

	// mytopic: since all
	messages, _ := s.Messages("mytopic", model.SinceAllMessages, false)
	require.Equal(t, 2, len(messages))
	require.Equal(t, "my message", messages[0].Message)
	require.Equal(t, "mytopic", messages[0].Topic)
	require.Equal(t, model.MessageEvent, messages[0].Event)
	require.Equal(t, "", messages[0].Title)
	require.Equal(t, 0, messages[0].Priority)
	require.Nil(t, messages[0].Tags)
	require.Equal(t, "my other message", messages[1].Message)

	// mytopic: since none
	messages, _ = s.Messages("mytopic", model.SinceNoMessages, false)
	require.Empty(t, messages)

	// mytopic: since m1 (by ID)
	messages, _ = s.Messages("mytopic", model.NewSinceID(m1.ID), false)
	require.Equal(t, 1, len(messages))
	require.Equal(t, m2.ID, messages[0].ID)
	require.Equal(t, "my other message", messages[0].Message)
	require.Equal(t, "mytopic", messages[0].Topic)

	// mytopic: since 2
	messages, _ = s.Messages("mytopic", model.NewSinceTime(2), false)
	require.Equal(t, 1, len(messages))
	require.Equal(t, "my other message", messages[0].Message)

	// mytopic: latest
	messages, _ = s.Messages("mytopic", model.SinceLatestMessage, false)
	require.Equal(t, 1, len(messages))
	require.Equal(t, "my other message", messages[0].Message)

	// example: count
	counts, err = s.MessageCounts()
	require.Nil(t, err)
	require.Equal(t, 1, counts["example"])

	// example: since all
	messages, _ = s.Messages("example", model.SinceAllMessages, false)
	require.Equal(t, "my example message", messages[0].Message)

	// non-existing: count
	counts, err = s.MessageCounts()
	require.Nil(t, err)
	require.Equal(t, 0, counts["doesnotexist"])

	// non-existing: since all
	messages, _ = s.Messages("doesnotexist", model.SinceAllMessages, false)
	require.Empty(t, messages)
}

func testCacheMessagesLock(t *testing.T, s message.Store) {
	var wg sync.WaitGroup
	for i := 0; i < 5000; i++ {
		wg.Add(1)
		go func() {
			assert.Nil(t, s.AddMessage(model.NewDefaultMessage("mytopic", "test message")))
			wg.Done()
		}()
	}
	wg.Wait()
}

func testCacheMessagesScheduled(t *testing.T, s message.Store) {
	m1 := model.NewDefaultMessage("mytopic", "message 1")
	m2 := model.NewDefaultMessage("mytopic", "message 2")
	m2.Time = time.Now().Add(time.Hour).Unix()
	m3 := model.NewDefaultMessage("mytopic", "message 3")
	m3.Time = time.Now().Add(time.Minute).Unix() // earlier than m2!
	m4 := model.NewDefaultMessage("mytopic2", "message 4")
	m4.Time = time.Now().Add(time.Minute).Unix()
	require.Nil(t, s.AddMessage(m1))
	require.Nil(t, s.AddMessage(m2))
	require.Nil(t, s.AddMessage(m3))

	messages, _ := s.Messages("mytopic", model.SinceAllMessages, false) // exclude scheduled
	require.Equal(t, 1, len(messages))
	require.Equal(t, "message 1", messages[0].Message)

	messages, _ = s.Messages("mytopic", model.SinceAllMessages, true) // include scheduled
	require.Equal(t, 3, len(messages))
	require.Equal(t, "message 1", messages[0].Message)
	require.Equal(t, "message 3", messages[1].Message) // Order!
	require.Equal(t, "message 2", messages[2].Message)

	messages, _ = s.MessagesDue()
	require.Empty(t, messages)
}

func testCacheTopics(t *testing.T, s message.Store) {
	require.Nil(t, s.AddMessage(model.NewDefaultMessage("topic1", "my example message")))
	require.Nil(t, s.AddMessage(model.NewDefaultMessage("topic2", "message 1")))
	require.Nil(t, s.AddMessage(model.NewDefaultMessage("topic2", "message 2")))
	require.Nil(t, s.AddMessage(model.NewDefaultMessage("topic2", "message 3")))

	topics, err := s.Topics()
	if err != nil {
		t.Fatal(err)
	}
	require.Equal(t, 2, len(topics))
	require.Contains(t, topics, "topic1")
	require.Contains(t, topics, "topic2")
}

func testCacheMessagesTagsPrioAndTitle(t *testing.T, s message.Store) {
	m := model.NewDefaultMessage("mytopic", "some message")
	m.Tags = []string{"tag1", "tag2"}
	m.Priority = 5
	m.Title = "some title"
	require.Nil(t, s.AddMessage(m))

	messages, _ := s.Messages("mytopic", model.SinceAllMessages, false)
	require.Equal(t, []string{"tag1", "tag2"}, messages[0].Tags)
	require.Equal(t, 5, messages[0].Priority)
	require.Equal(t, "some title", messages[0].Title)
}

func testCacheMessagesSinceID(t *testing.T, s message.Store) {
	m1 := model.NewDefaultMessage("mytopic", "message 1")
	m1.Time = 100
	m2 := model.NewDefaultMessage("mytopic", "message 2")
	m2.Time = 200
	m3 := model.NewDefaultMessage("mytopic", "message 3")
	m3.Time = time.Now().Add(time.Hour).Unix() // Scheduled, in the future, later than m7 and m5
	m4 := model.NewDefaultMessage("mytopic", "message 4")
	m4.Time = 400
	m5 := model.NewDefaultMessage("mytopic", "message 5")
	m5.Time = time.Now().Add(time.Minute).Unix() // Scheduled, in the future, later than m7
	m6 := model.NewDefaultMessage("mytopic", "message 6")
	m6.Time = 600
	m7 := model.NewDefaultMessage("mytopic", "message 7")
	m7.Time = 700

	require.Nil(t, s.AddMessage(m1))
	require.Nil(t, s.AddMessage(m2))
	require.Nil(t, s.AddMessage(m3))
	require.Nil(t, s.AddMessage(m4))
	require.Nil(t, s.AddMessage(m5))
	require.Nil(t, s.AddMessage(m6))
	require.Nil(t, s.AddMessage(m7))

	// Case 1: Since ID exists, exclude scheduled
	messages, _ := s.Messages("mytopic", model.NewSinceID(m2.ID), false)
	require.Equal(t, 3, len(messages))
	require.Equal(t, "message 4", messages[0].Message)
	require.Equal(t, "message 6", messages[1].Message) // Not scheduled m3/m5!
	require.Equal(t, "message 7", messages[2].Message)

	// Case 2: Since ID exists, include scheduled
	messages, _ = s.Messages("mytopic", model.NewSinceID(m2.ID), true)
	require.Equal(t, 5, len(messages))
	require.Equal(t, "message 4", messages[0].Message)
	require.Equal(t, "message 6", messages[1].Message)
	require.Equal(t, "message 7", messages[2].Message)
	require.Equal(t, "message 5", messages[3].Message) // Order!
	require.Equal(t, "message 3", messages[4].Message) // Order!

	// Case 3: Since ID does not exist (-> Return all messages), include scheduled
	messages, _ = s.Messages("mytopic", model.NewSinceID("doesntexist"), true)
	require.Equal(t, 7, len(messages))
	require.Equal(t, "message 1", messages[0].Message)
	require.Equal(t, "message 2", messages[1].Message)
	require.Equal(t, "message 4", messages[2].Message)
	require.Equal(t, "message 6", messages[3].Message)
	require.Equal(t, "message 7", messages[4].Message)
	require.Equal(t, "message 5", messages[5].Message) // Order!
	require.Equal(t, "message 3", messages[6].Message) // Order!

	// Case 4: Since ID exists and is last message (-> Return no messages), exclude scheduled
	messages, _ = s.Messages("mytopic", model.NewSinceID(m7.ID), false)
	require.Equal(t, 0, len(messages))

	// Case 5: Since ID exists and is last message (-> Return no messages), include scheduled
	messages, _ = s.Messages("mytopic", model.NewSinceID(m7.ID), true)
	require.Equal(t, 2, len(messages))
	require.Equal(t, "message 5", messages[0].Message)
	require.Equal(t, "message 3", messages[1].Message)
}

func testCachePrune(t *testing.T, s message.Store) {
	now := time.Now().Unix()

	m1 := model.NewDefaultMessage("mytopic", "my message")
	m1.Time = now - 10
	m1.Expires = now - 5

	m2 := model.NewDefaultMessage("mytopic", "my other message")
	m2.Time = now - 5
	m2.Expires = now + 5 // In the future

	m3 := model.NewDefaultMessage("another_topic", "and another one")
	m3.Time = now - 12
	m3.Expires = now - 2

	require.Nil(t, s.AddMessage(m1))
	require.Nil(t, s.AddMessage(m2))
	require.Nil(t, s.AddMessage(m3))

	counts, err := s.MessageCounts()
	require.Nil(t, err)
	require.Equal(t, 2, counts["mytopic"])
	require.Equal(t, 1, counts["another_topic"])

	expiredMessageIDs, err := s.MessagesExpired()
	require.Nil(t, err)
	require.Nil(t, s.DeleteMessages(expiredMessageIDs...))

	counts, err = s.MessageCounts()
	require.Nil(t, err)
	require.Equal(t, 1, counts["mytopic"])
	require.Equal(t, 0, counts["another_topic"])

	messages, err := s.Messages("mytopic", model.SinceAllMessages, false)
	require.Nil(t, err)
	require.Equal(t, 1, len(messages))
	require.Equal(t, "my other message", messages[0].Message)
}

func testCacheAttachments(t *testing.T, s message.Store) {
	expires1 := time.Now().Add(-4 * time.Hour).Unix() // Expired
	m := model.NewDefaultMessage("mytopic", "flower for you")
	m.ID = "m1"
	m.SequenceID = "m1"
	m.Sender = netip.MustParseAddr("1.2.3.4")
	m.Attachment = &model.Attachment{
		Name:    "flower.jpg",
		Type:    "image/jpeg",
		Size:    5000,
		Expires: expires1,
		URL:     "https://ntfy.sh/file/AbDeFgJhal.jpg",
	}
	require.Nil(t, s.AddMessage(m))

	expires2 := time.Now().Add(2 * time.Hour).Unix() // Future
	m = model.NewDefaultMessage("mytopic", "sending you a car")
	m.ID = "m2"
	m.SequenceID = "m2"
	m.Sender = netip.MustParseAddr("1.2.3.4")
	m.Attachment = &model.Attachment{
		Name:    "car.jpg",
		Type:    "image/jpeg",
		Size:    10000,
		Expires: expires2,
		URL:     "https://ntfy.sh/file/aCaRURL.jpg",
	}
	require.Nil(t, s.AddMessage(m))

	expires3 := time.Now().Add(1 * time.Hour).Unix() // Future
	m = model.NewDefaultMessage("another-topic", "sending you another car")
	m.ID = "m3"
	m.SequenceID = "m3"
	m.User = "u_BAsbaAa"
	m.Sender = netip.MustParseAddr("5.6.7.8")
	m.Attachment = &model.Attachment{
		Name:    "another-car.jpg",
		Type:    "image/jpeg",
		Size:    20000,
		Expires: expires3,
		URL:     "https://ntfy.sh/file/zakaDHFW.jpg",
	}
	require.Nil(t, s.AddMessage(m))

	messages, err := s.Messages("mytopic", model.SinceAllMessages, false)
	require.Nil(t, err)
	require.Equal(t, 2, len(messages))

	require.Equal(t, "flower for you", messages[0].Message)
	require.Equal(t, "flower.jpg", messages[0].Attachment.Name)
	require.Equal(t, "image/jpeg", messages[0].Attachment.Type)
	require.Equal(t, int64(5000), messages[0].Attachment.Size)
	require.Equal(t, expires1, messages[0].Attachment.Expires)
	require.Equal(t, "https://ntfy.sh/file/AbDeFgJhal.jpg", messages[0].Attachment.URL)
	require.Equal(t, "1.2.3.4", messages[0].Sender.String())

	require.Equal(t, "sending you a car", messages[1].Message)
	require.Equal(t, "car.jpg", messages[1].Attachment.Name)
	require.Equal(t, "image/jpeg", messages[1].Attachment.Type)
	require.Equal(t, int64(10000), messages[1].Attachment.Size)
	require.Equal(t, expires2, messages[1].Attachment.Expires)
	require.Equal(t, "https://ntfy.sh/file/aCaRURL.jpg", messages[1].Attachment.URL)
	require.Equal(t, "1.2.3.4", messages[1].Sender.String())

	size, err := s.AttachmentBytesUsedBySender("1.2.3.4")
	require.Nil(t, err)
	require.Equal(t, int64(10000), size)

	size, err = s.AttachmentBytesUsedBySender("5.6.7.8")
	require.Nil(t, err)
	require.Equal(t, int64(0), size) // Accounted to the user, not the IP!

	size, err = s.AttachmentBytesUsedByUser("u_BAsbaAa")
	require.Nil(t, err)
	require.Equal(t, int64(20000), size)
}

func testCacheAttachmentsExpired(t *testing.T, s message.Store) {
	m := model.NewDefaultMessage("mytopic", "flower for you")
	m.ID = "m1"
	m.SequenceID = "m1"
	m.Expires = time.Now().Add(time.Hour).Unix()
	require.Nil(t, s.AddMessage(m))

	m = model.NewDefaultMessage("mytopic", "message with attachment")
	m.ID = "m2"
	m.SequenceID = "m2"
	m.Expires = time.Now().Add(2 * time.Hour).Unix()
	m.Attachment = &model.Attachment{
		Name:    "car.jpg",
		Type:    "image/jpeg",
		Size:    10000,
		Expires: time.Now().Add(2 * time.Hour).Unix(),
		URL:     "https://ntfy.sh/file/aCaRURL.jpg",
	}
	require.Nil(t, s.AddMessage(m))

	m = model.NewDefaultMessage("mytopic", "message with external attachment")
	m.ID = "m3"
	m.SequenceID = "m3"
	m.Expires = time.Now().Add(2 * time.Hour).Unix()
	m.Attachment = &model.Attachment{
		Name:    "car.jpg",
		Type:    "image/jpeg",
		Expires: 0, // Unknown!
		URL:     "https://somedomain.com/car.jpg",
	}
	require.Nil(t, s.AddMessage(m))

	m = model.NewDefaultMessage("mytopic2", "message with expired attachment")
	m.ID = "m4"
	m.SequenceID = "m4"
	m.Expires = time.Now().Add(2 * time.Hour).Unix()
	m.Attachment = &model.Attachment{
		Name:    "expired-car.jpg",
		Type:    "image/jpeg",
		Size:    20000,
		Expires: time.Now().Add(-1 * time.Hour).Unix(),
		URL:     "https://ntfy.sh/file/aCaRURL.jpg",
	}
	require.Nil(t, s.AddMessage(m))

	ids, err := s.AttachmentsExpired()
	require.Nil(t, err)
	require.Equal(t, 1, len(ids))
	require.Equal(t, "m4", ids[0])
}

func testSender(t *testing.T, s message.Store) {
	m1 := model.NewDefaultMessage("mytopic", "mymessage")
	m1.Sender = netip.MustParseAddr("1.2.3.4")
	require.Nil(t, s.AddMessage(m1))

	m2 := model.NewDefaultMessage("mytopic", "mymessage without sender")
	require.Nil(t, s.AddMessage(m2))

	messages, err := s.Messages("mytopic", model.SinceAllMessages, false)
	require.Nil(t, err)
	require.Equal(t, 2, len(messages))
	require.Equal(t, messages[0].Sender, netip.MustParseAddr("1.2.3.4"))
	require.Equal(t, messages[1].Sender, netip.Addr{})
}

func testDeleteScheduledBySequenceID(t *testing.T, s message.Store) {
	// Create a scheduled (unpublished) message
	scheduledMsg := model.NewDefaultMessage("mytopic", "scheduled message")
	scheduledMsg.ID = "scheduled1"
	scheduledMsg.SequenceID = "seq123"
	scheduledMsg.Time = time.Now().Add(time.Hour).Unix() // Future time makes it scheduled
	require.Nil(t, s.AddMessage(scheduledMsg))

	// Create a published message with different sequence ID
	publishedMsg := model.NewDefaultMessage("mytopic", "published message")
	publishedMsg.ID = "published1"
	publishedMsg.SequenceID = "seq456"
	publishedMsg.Time = time.Now().Add(-time.Hour).Unix() // Past time makes it published
	require.Nil(t, s.AddMessage(publishedMsg))

	// Create a scheduled message in a different topic
	otherTopicMsg := model.NewDefaultMessage("othertopic", "other scheduled")
	otherTopicMsg.ID = "other1"
	otherTopicMsg.SequenceID = "seq123" // Same sequence ID as scheduledMsg
	otherTopicMsg.Time = time.Now().Add(time.Hour).Unix()
	require.Nil(t, s.AddMessage(otherTopicMsg))

	// Verify all messages exist (including scheduled)
	messages, err := s.Messages("mytopic", model.SinceAllMessages, true)
	require.Nil(t, err)
	require.Equal(t, 2, len(messages))

	messages, err = s.Messages("othertopic", model.SinceAllMessages, true)
	require.Nil(t, err)
	require.Equal(t, 1, len(messages))

	// Delete scheduled message by sequence ID and verify returned IDs
	deletedIDs, err := s.DeleteScheduledBySequenceID("mytopic", "seq123")
	require.Nil(t, err)
	require.Equal(t, 1, len(deletedIDs))
	require.Equal(t, "scheduled1", deletedIDs[0])

	// Verify scheduled message is deleted
	messages, err = s.Messages("mytopic", model.SinceAllMessages, true)
	require.Nil(t, err)
	require.Equal(t, 1, len(messages))
	require.Equal(t, "published message", messages[0].Message)

	// Verify other topic's message still exists (topic-scoped deletion)
	messages, err = s.Messages("othertopic", model.SinceAllMessages, true)
	require.Nil(t, err)
	require.Equal(t, 1, len(messages))
	require.Equal(t, "other scheduled", messages[0].Message)

	// Deleting non-existent sequence ID should return empty list
	deletedIDs, err = s.DeleteScheduledBySequenceID("mytopic", "nonexistent")
	require.Nil(t, err)
	require.Empty(t, deletedIDs)

	// Deleting published message should not affect it (only deletes unpublished)
	deletedIDs, err = s.DeleteScheduledBySequenceID("mytopic", "seq456")
	require.Nil(t, err)
	require.Empty(t, deletedIDs)

	messages, err = s.Messages("mytopic", model.SinceAllMessages, true)
	require.Nil(t, err)
	require.Equal(t, 1, len(messages))
	require.Equal(t, "published message", messages[0].Message)
}

func testMessageByID(t *testing.T, s message.Store) {
	// Add a message
	m := model.NewDefaultMessage("mytopic", "some message")
	m.Title = "some title"
	m.Priority = 4
	m.Tags = []string{"tag1", "tag2"}
	require.Nil(t, s.AddMessage(m))

	// Retrieve by ID
	retrieved, err := s.Message(m.ID)
	require.Nil(t, err)
	require.Equal(t, m.ID, retrieved.ID)
	require.Equal(t, "mytopic", retrieved.Topic)
	require.Equal(t, "some message", retrieved.Message)
	require.Equal(t, "some title", retrieved.Title)
	require.Equal(t, 4, retrieved.Priority)
	require.Equal(t, []string{"tag1", "tag2"}, retrieved.Tags)

	// Non-existent ID returns ErrMessageNotFound
	_, err = s.Message("doesnotexist")
	require.Equal(t, model.ErrMessageNotFound, err)
}

func testMarkPublished(t *testing.T, s message.Store) {
	// Add a scheduled message (future time → unpublished)
	m := model.NewDefaultMessage("mytopic", "scheduled message")
	m.Time = time.Now().Add(time.Hour).Unix()
	require.Nil(t, s.AddMessage(m))

	// Verify it does not appear in non-scheduled queries
	messages, err := s.Messages("mytopic", model.SinceAllMessages, false)
	require.Nil(t, err)
	require.Equal(t, 0, len(messages))

	// Verify it does appear in scheduled queries
	messages, err = s.Messages("mytopic", model.SinceAllMessages, true)
	require.Nil(t, err)
	require.Equal(t, 1, len(messages))

	// Mark as published
	require.Nil(t, s.MarkPublished(m))

	// Now it should appear in non-scheduled queries too
	messages, err = s.Messages("mytopic", model.SinceAllMessages, false)
	require.Nil(t, err)
	require.Equal(t, 1, len(messages))
	require.Equal(t, "scheduled message", messages[0].Message)
}

func testExpireMessages(t *testing.T, s message.Store) {
	// Add messages to two topics
	m1 := model.NewDefaultMessage("topic1", "message 1")
	m1.Expires = time.Now().Add(time.Hour).Unix()
	m2 := model.NewDefaultMessage("topic1", "message 2")
	m2.Expires = time.Now().Add(time.Hour).Unix()
	m3 := model.NewDefaultMessage("topic2", "message 3")
	m3.Expires = time.Now().Add(time.Hour).Unix()
	require.Nil(t, s.AddMessage(m1))
	require.Nil(t, s.AddMessage(m2))
	require.Nil(t, s.AddMessage(m3))

	// Verify all messages exist
	messages, err := s.Messages("topic1", model.SinceAllMessages, false)
	require.Nil(t, err)
	require.Equal(t, 2, len(messages))
	messages, err = s.Messages("topic2", model.SinceAllMessages, false)
	require.Nil(t, err)
	require.Equal(t, 1, len(messages))

	// Expire topic1 messages
	require.Nil(t, s.ExpireMessages("topic1"))

	// topic1 messages should now be expired (expires set to past)
	expiredIDs, err := s.MessagesExpired()
	require.Nil(t, err)
	require.Equal(t, 2, len(expiredIDs))
	sort.Strings(expiredIDs)
	expectedIDs := []string{m1.ID, m2.ID}
	sort.Strings(expectedIDs)
	require.Equal(t, expectedIDs, expiredIDs)

	// topic2 should be unaffected
	messages, err = s.Messages("topic2", model.SinceAllMessages, false)
	require.Nil(t, err)
	require.Equal(t, 1, len(messages))
	require.Equal(t, "message 3", messages[0].Message)
}

func testMarkAttachmentsDeleted(t *testing.T, s message.Store) {
	// Add a message with an expired attachment (file needs cleanup)
	m1 := model.NewDefaultMessage("mytopic", "old file")
	m1.ID = "msg1"
	m1.SequenceID = "msg1"
	m1.Expires = time.Now().Add(time.Hour).Unix()
	m1.Attachment = &model.Attachment{
		Name:    "old.pdf",
		Type:    "application/pdf",
		Size:    50000,
		Expires: time.Now().Add(-time.Hour).Unix(), // Expired
		URL:     "https://ntfy.sh/file/old.pdf",
	}
	require.Nil(t, s.AddMessage(m1))

	// Add a message with another expired attachment
	m2 := model.NewDefaultMessage("mytopic", "another old file")
	m2.ID = "msg2"
	m2.SequenceID = "msg2"
	m2.Expires = time.Now().Add(time.Hour).Unix()
	m2.Attachment = &model.Attachment{
		Name:    "another.pdf",
		Type:    "application/pdf",
		Size:    30000,
		Expires: time.Now().Add(-time.Hour).Unix(), // Expired
		URL:     "https://ntfy.sh/file/another.pdf",
	}
	require.Nil(t, s.AddMessage(m2))

	// Both should show as expired attachments needing cleanup
	ids, err := s.AttachmentsExpired()
	require.Nil(t, err)
	require.Equal(t, 2, len(ids))

	// Mark msg1's attachment as deleted (file cleaned up)
	require.Nil(t, s.MarkAttachmentsDeleted("msg1"))

	// Now only msg2 should show as needing cleanup
	ids, err = s.AttachmentsExpired()
	require.Nil(t, err)
	require.Equal(t, 1, len(ids))
	require.Equal(t, "msg2", ids[0])

	// Mark msg2 too
	require.Nil(t, s.MarkAttachmentsDeleted("msg2"))

	// No more expired attachments to clean up
	ids, err = s.AttachmentsExpired()
	require.Nil(t, err)
	require.Equal(t, 0, len(ids))

	// Messages themselves still exist
	messages, err := s.Messages("mytopic", model.SinceAllMessages, false)
	require.Nil(t, err)
	require.Equal(t, 2, len(messages))
}

func testStats(t *testing.T, s message.Store) {
	// Initial stats should be zero
	messages, err := s.Stats()
	require.Nil(t, err)
	require.Equal(t, int64(0), messages)

	// Update stats
	require.Nil(t, s.UpdateStats(42))
	messages, err = s.Stats()
	require.Nil(t, err)
	require.Equal(t, int64(42), messages)

	// Update again (overwrites)
	require.Nil(t, s.UpdateStats(100))
	messages, err = s.Stats()
	require.Nil(t, err)
	require.Equal(t, int64(100), messages)
}

func testAddMessages(t *testing.T, s message.Store) {
	// Batch add multiple messages
	msgs := []*model.Message{
		model.NewDefaultMessage("mytopic", "batch 1"),
		model.NewDefaultMessage("mytopic", "batch 2"),
		model.NewDefaultMessage("othertopic", "batch 3"),
	}
	require.Nil(t, s.AddMessages(msgs))

	// Verify all were inserted
	messages, err := s.Messages("mytopic", model.SinceAllMessages, false)
	require.Nil(t, err)
	require.Equal(t, 2, len(messages))

	messages, err = s.Messages("othertopic", model.SinceAllMessages, false)
	require.Nil(t, err)
	require.Equal(t, 1, len(messages))
	require.Equal(t, "batch 3", messages[0].Message)

	// Empty batch should succeed
	require.Nil(t, s.AddMessages([]*model.Message{}))

	// Batch with invalid event type should fail
	badMsgs := []*model.Message{
		model.NewKeepaliveMessage("mytopic"),
	}
	require.NotNil(t, s.AddMessages(badMsgs))
}

func testMessagesDue(t *testing.T, s message.Store) {
	// Add a message scheduled in the past (i.e. it's due now)
	m1 := model.NewDefaultMessage("mytopic", "due message")
	m1.Time = time.Now().Add(-time.Second).Unix()
	// Set expires in the future so it doesn't get pruned
	m1.Expires = time.Now().Add(time.Hour).Unix()
	require.Nil(t, s.AddMessage(m1))

	// Add a message scheduled in the future (not due)
	m2 := model.NewDefaultMessage("mytopic", "future message")
	m2.Time = time.Now().Add(time.Hour).Unix()
	require.Nil(t, s.AddMessage(m2))

	// Mark m1 as published so it won't be "due"
	// (MessagesDue returns unpublished messages whose time <= now)
	// m1 is auto-published (time <= now), so it should not be due
	// m2 is unpublished (time in future), not due yet
	due, err := s.MessagesDue()
	require.Nil(t, err)
	require.Equal(t, 0, len(due))

	// Add a message that was explicitly scheduled in the past but time has "arrived"
	// We need to manipulate the database to create a truly "due" message:
	// a message with published=false and time <= now
	m3 := model.NewDefaultMessage("mytopic", "truly due message")
	m3.Time = time.Now().Add(2 * time.Second).Unix() // 2 seconds from now
	require.Nil(t, s.AddMessage(m3))

	// Not due yet
	due, err = s.MessagesDue()
	require.Nil(t, err)
	require.Equal(t, 0, len(due))

	// Wait for it to become due
	time.Sleep(3 * time.Second)

	due, err = s.MessagesDue()
	require.Nil(t, err)
	require.Equal(t, 1, len(due))
	require.Equal(t, "truly due message", due[0].Message)
}

func testMessageFieldRoundTrip(t *testing.T, s message.Store) {
	// Create a message with all fields populated
	m := model.NewDefaultMessage("mytopic", "hello world")
	m.SequenceID = "custom_seq_id"
	m.Title = "A Title"
	m.Priority = 4
	m.Tags = []string{"warning", "srv01"}
	m.Click = "https://example.com/click"
	m.Icon = "https://example.com/icon.png"
	m.Actions = []*model.Action{
		{
			ID:     "action1",
			Action: "view",
			Label:  "Open Site",
			URL:    "https://example.com",
			Clear:  true,
		},
		{
			ID:      "action2",
			Action:  "http",
			Label:   "Call Webhook",
			URL:     "https://example.com/hook",
			Method:  "PUT",
			Headers: map[string]string{"X-Token": "secret"},
			Body:    `{"key":"value"}`,
		},
	}
	m.ContentType = "text/markdown"
	m.Encoding = "base64"
	m.Sender = netip.MustParseAddr("9.8.7.6")
	m.User = "u_TestUser123"
	require.Nil(t, s.AddMessage(m))

	// Retrieve and verify every field
	retrieved, err := s.Message(m.ID)
	require.Nil(t, err)

	require.Equal(t, m.ID, retrieved.ID)
	require.Equal(t, "custom_seq_id", retrieved.SequenceID)
	require.Equal(t, m.Time, retrieved.Time)
	require.Equal(t, m.Expires, retrieved.Expires)
	require.Equal(t, model.MessageEvent, retrieved.Event)
	require.Equal(t, "mytopic", retrieved.Topic)
	require.Equal(t, "hello world", retrieved.Message)
	require.Equal(t, "A Title", retrieved.Title)
	require.Equal(t, 4, retrieved.Priority)
	require.Equal(t, []string{"warning", "srv01"}, retrieved.Tags)
	require.Equal(t, "https://example.com/click", retrieved.Click)
	require.Equal(t, "https://example.com/icon.png", retrieved.Icon)
	require.Equal(t, "text/markdown", retrieved.ContentType)
	require.Equal(t, "base64", retrieved.Encoding)
	require.Equal(t, netip.MustParseAddr("9.8.7.6"), retrieved.Sender)
	require.Equal(t, "u_TestUser123", retrieved.User)

	// Verify actions round-trip
	require.Equal(t, 2, len(retrieved.Actions))

	require.Equal(t, "action1", retrieved.Actions[0].ID)
	require.Equal(t, "view", retrieved.Actions[0].Action)
	require.Equal(t, "Open Site", retrieved.Actions[0].Label)
	require.Equal(t, "https://example.com", retrieved.Actions[0].URL)
	require.Equal(t, true, retrieved.Actions[0].Clear)

	require.Equal(t, "action2", retrieved.Actions[1].ID)
	require.Equal(t, "http", retrieved.Actions[1].Action)
	require.Equal(t, "Call Webhook", retrieved.Actions[1].Label)
	require.Equal(t, "https://example.com/hook", retrieved.Actions[1].URL)
	require.Equal(t, "PUT", retrieved.Actions[1].Method)
	require.Equal(t, "secret", retrieved.Actions[1].Headers["X-Token"])
	require.Equal(t, `{"key":"value"}`, retrieved.Actions[1].Body)
}