m8sh/database/leveldb/subscriptions_test.go

package leveldb

import (
	"testing"
	"time"

	"github.com/d1nch8g/mesh/database"
	"github.com/google/uuid"
	"github.com/stretchr/testify/require"
)

func TestSubscriptions_Subscribe(t *testing.T) {
	db, cleanup := openDB(t)
	defer cleanup()

	subs := db.Subscriptions()

	t.Run("subscribe new target", func(t *testing.T) {
		err := subs.Subscribe("masha@d1.com", "bob@d2.io", []byte("sig123"))
		require.NoError(t, err)
	})

	t.Run("subscribe overwrites existing", func(t *testing.T) {
		err := subs.Subscribe("masha@d1.com", "bob@d2.io", []byte("sig456"))
		require.NoError(t, err)
	})

	t.Run("subscribe multiple targets", func(t *testing.T) {
		require.NoError(t, subs.Subscribe("masha@d1.com", "alice@d3.net", []byte("sig789")))
		require.NoError(t, subs.Subscribe("masha@d1.com", "carol@d4.org", []byte("sig000")))
	})

	t.Run("subscribe different users", func(t *testing.T) {
		require.NoError(t, subs.Subscribe("petya@d1.com", "bob@d2.io", []byte("sig111")))
		require.NoError(t, subs.Subscribe("petya@d1.com", "alice@d3.net", []byte("sig222")))
	})

	t.Run("subscribe empty signature", func(t *testing.T) {
		err := subs.Subscribe("masha@d1.com", "empty-sig@d5.io", nil)
		require.NoError(t, err)
	})
}

func TestSubscriptions_Unsubscribe(t *testing.T) {
	db, cleanup := openDB(t)
	defer cleanup()

	subs := db.Subscriptions()

	require.NoError(t, subs.Subscribe("masha@d1.com", "bob@d2.io", []byte("sig123")))
	require.NoError(t, subs.Subscribe("masha@d1.com", "alice@d3.net", []byte("sig456")))

	t.Run("unsubscribe existing", func(t *testing.T) {
		err := subs.Unsubscribe("masha@d1.com", "bob@d2.io")
		require.NoError(t, err)

		list, err := subs.OfUser("masha@d1.com")
		require.NoError(t, err)
		for _, sub := range list {
			require.NotEqual(t, "bob@d2.io", sub.Email)
		}
	})

	t.Run("unsubscribe non-existing returns ErrNotFound", func(t *testing.T) {
		err := subs.Unsubscribe("masha@d1.com", "nonexistent@d5.io")
		require.ErrorIs(t, err, database.ErrNotFound)
	})

	t.Run("unsubscribe already removed returns ErrNotFound", func(t *testing.T) {
		err := subs.Unsubscribe("masha@d1.com", "bob@d2.io")
		require.ErrorIs(t, err, database.ErrNotFound)
	})
}

func TestSubscriptions_OfUser(t *testing.T) {
	db, cleanup := openDB(t)
	defer cleanup()

	subs := db.Subscriptions()

	t.Run("empty subscriptions", func(t *testing.T) {
		list, err := subs.OfUser("newuser@d1.com")
		require.NoError(t, err)
		require.Empty(t, list)
	})

	t.Run("single subscription", func(t *testing.T) {
		require.NoError(t, subs.Subscribe("masha@d1.com", "bob@d2.io", []byte("sig123")))

		list, err := subs.OfUser("masha@d1.com")
		require.NoError(t, err)
		require.Len(t, list, 1)
		require.Equal(t, "bob@d2.io", list[0].Email)
		require.Equal(t, []byte("sig123"), list[0].Signature)
	})

	t.Run("multiple subscriptions", func(t *testing.T) {
		require.NoError(t, subs.Subscribe("masha@d1.com", "alice@d3.net", []byte("sig456")))

		list, err := subs.OfUser("masha@d1.com")
		require.NoError(t, err)
		require.Len(t, list, 2)
	})

	t.Run("isolation between users", func(t *testing.T) {
		require.NoError(t, subs.Subscribe("petya@d1.com", "bob@d2.io", []byte("sig777")))

		mashaList, err := subs.OfUser("masha@d1.com")
		require.NoError(t, err)

		petyaList, err := subs.OfUser("petya@d1.com")
		require.NoError(t, err)

		require.Len(t, mashaList, 2)
		require.Len(t, petyaList, 1)
	})
}

func TestSubscriptions_ToUser(t *testing.T) {
	db, cleanup := openDB(t)
	defer cleanup()

	subs := db.Subscriptions()

	t.Run("empty subscribers", func(t *testing.T) {
		list, err := subs.ToUser("bob@d2.io")
		require.NoError(t, err)
		require.Empty(t, list)
	})
}

func TestSubscriptions_UpdateReference(t *testing.T) {
	db, cleanup := openDB(t)
	defer cleanup()
	db.subscriptions.flushInterval = 10 * time.Millisecond

	subs := db.Subscriptions()

	t.Run("increment new domain", func(t *testing.T) {
		err := subs.UpdateReference("bob@d2.io", "d1.com", 1)
		require.NoError(t, err)
	})

	t.Run("increment existing domain multiple times", func(t *testing.T) {
		require.NoError(t, subs.UpdateReference("bob@d2.io", "d1.com", 2))
		require.NoError(t, subs.UpdateReference("bob@d2.io", "d1.com", 3))
	})

	t.Run("decrement domain partial", func(t *testing.T) {
		require.NoError(t, subs.UpdateReference("bob@d2.io", "d1.com", -2))
	})

	t.Run("decrement to zero deletes key", func(t *testing.T) {
		user := uuid.NewString()
		require.NoError(t, subs.UpdateReference(user, "d2.io", 1))
		require.NoError(t, subs.UpdateReference(user, "d2.io", -1))
		time.Sleep(50 * time.Millisecond)
		refs, err := subs.ListReferences(user)
		require.NoError(t, err)
		for _, ref := range refs {
			require.NotEqual(t, "d2.io", ref.Domain)
		}
	})

	t.Run("decrement below zero clamps to zero", func(t *testing.T) {
		require.NoError(t, subs.UpdateReference("bob@d2.io", "d3.net", 1))
		require.NoError(t, subs.UpdateReference("bob@d2.io", "d3.net", -10))

		time.Sleep(50 * time.Millisecond)

		refs, err := subs.ListReferences("bob@d2.io")
		require.NoError(t, err)
		for _, ref := range refs {
			require.NotEqual(t, "d3.net", ref.Domain)
		}
	})

	t.Run("negative delta only", func(t *testing.T) {
		err := subs.UpdateReference("bob@d2.io", "d4.org", -3)
		require.NoError(t, err)
	})

	t.Run("update total counter consistency", func(t *testing.T) {
		require.NoError(t, db.Users().Create(database.User{Name: "charlie@d5.io"}))

		require.NoError(t, subs.UpdateReference("charlie@d5.io", "d1.com", 5))
		require.NoError(t, subs.UpdateReference("charlie@d5.io", "d2.io", 3))
		require.NoError(t, subs.UpdateReference("charlie@d5.io", "d1.com", -2))

		time.Sleep(50 * time.Millisecond)

		user, err := db.Users().Get("charlie@d5.io")
		require.NoError(t, err)
		require.Equal(t, uint64(6), user.Subscribers)
	})
}

func TestSubscriptions_ListReferences(t *testing.T) {
	db, cleanup := openDB(t)
	defer cleanup()
	db.subscriptions.flushInterval = 10 * time.Millisecond

	subs := db.Subscriptions()

	t.Run("empty references", func(t *testing.T) {
		refs, err := subs.ListReferences("unknown@d1.com")
		require.NoError(t, err)
		require.Empty(t, refs)
	})

	t.Run("single reference", func(t *testing.T) {
		err := subs.UpdateReference("ref-user@x.com", "d1.com", 5)
		require.NoError(t, err)

		time.Sleep(50 * time.Millisecond)

		refs, err := subs.ListReferences("ref-user@x.com")
		require.NoError(t, err)
		require.Len(t, refs, 1)
		require.Equal(t, "d1.com", refs[0].Domain)
		require.Equal(t, uint64(5), refs[0].Count)
	})

	t.Run("multiple references sorted", func(t *testing.T) {
		require.NoError(t, subs.UpdateReference("alice@d3.net", "d1.com", 3))
		require.NoError(t, subs.UpdateReference("alice@d3.net", "d2.io", 7))

		time.Sleep(50 * time.Millisecond)

		refs, err := subs.ListReferences("alice@d3.net")
		require.NoError(t, err)
		require.Len(t, refs, 2)
	})

	t.Run("deleted domain not in list", func(t *testing.T) {
		require.NoError(t, subs.UpdateReference("carol@d4.org", "temp.io", 1))
		require.NoError(t, subs.UpdateReference("carol@d4.org", "temp.io", -1))

		time.Sleep(50 * time.Millisecond)

		refs, err := subs.ListReferences("carol@d4.org")
		require.NoError(t, err)
		for _, ref := range refs {
			require.NotEqual(t, "temp.io", ref.Domain)
		}
	})
}

func TestSubscriptions_AfterCloseReturnsError(t *testing.T) {
	path := t.TempDir() + "/db"
	db, err := New(path)
	require.NoError(t, err)

	subs := db.Subscriptions()
	require.NoError(t, subs.Subscribe("masha@d1.com", "bob@d2.io", []byte("sig")))
	require.NoError(t, db.Close())

	// After close, operations may error or return empty — just verify no panic.
	require.NotPanics(t, func() {
		subs.OfUser("masha@d1.com")
		subs.Subscribe("masha@d1.com", "carol@d4.org", []byte("sig"))
		subs.Unsubscribe("masha@d1.com", "bob@d2.io")
		subs.ListReferences("masha@d1.com")
	})
}

func TestSubscriptions_ToUser_Integration(t *testing.T) {
	db, cleanup := openDB(t)
	defer cleanup()

	subs := db.Subscriptions()

	require.NoError(t, subs.Subscribe("masha@d1.com", "bob@d2.io", []byte("sig-masha-bob")))
	require.NoError(t, subs.Subscribe("petya@d1.com", "bob@d2.io", []byte("sig-petya-bob")))
	require.NoError(t, subs.Subscribe("alice@d1.com", "bob@d2.io", []byte("sig-alice-bob")))

	list, err := subs.ToUser("bob@d2.io")
	require.NoError(t, err)
	require.Len(t, list, 3)

	emails := make(map[string]bool)
	for _, sub := range list {
		emails[sub.Email] = true
	}
	require.True(t, emails["masha@d1.com"])
	require.True(t, emails["petya@d1.com"])
	require.True(t, emails["alice@d1.com"])
}