This commit is contained in:
KS Jannette
2026-03-28 11:55:51 -04:00
parent d9c3bd1db5
commit b0572451d3
18 changed files with 293 additions and 32 deletions

Binary file not shown.

View File

@@ -51,6 +51,7 @@ func main() {
defer database.Close()
userModel := models.NewUserModel(pool)
addressModel := models.NewAddressModel(pool)
alertRuleModel := models.NewAlertRuleModel(pool)
alertEventModel := models.NewAlertEventModel(pool)
@@ -59,7 +60,7 @@ func main() {
observer := services.NewObserverService(eth, addressModel, checkpointModel)
evaluator := services.NewEvaluatorService(
eth, alertRuleModel, alertEventModel, addressModel, notifConfigModel,
eth, alertRuleModel, alertEventModel, addressModel, userModel, notifConfigModel,
)
digestSvc := services.NewEmailDigestService(cfg.ResendAPIKey, cfg.EmailFrom, alertEventModel, notifConfigModel)

View File

@@ -94,6 +94,19 @@ func (c *Config) DSN() string {
)
}
// TierForPriceID maps a Stripe price ID back to the corresponding
// subscription tier. Returns empty string if the price is unrecognised.
func (c *Config) TierForPriceID(priceID string) string {
switch priceID {
case c.StripePriceIDPremium:
return "premium"
case c.StripePriceIDPro:
return "pro"
default:
return ""
}
}
func getEnv(key, fallback string) string {
if v := os.Getenv(key); v != "" {
return v

View File

@@ -0,0 +1,28 @@
package config
import "testing"
func TestTierForPriceID(t *testing.T) {
t.Parallel()
cfg := &Config{
StripePriceIDPremium: "price_premium_123",
StripePriceIDPro: "price_pro_456",
}
tests := []struct {
priceID string
want string
}{
{"price_premium_123", "premium"},
{"price_pro_456", "pro"},
{"price_unknown", ""},
{"", ""},
}
for _, tt := range tests {
if got := cfg.TierForPriceID(tt.priceID); got != tt.want {
t.Errorf("TierForPriceID(%q) = %q, want %q", tt.priceID, got, tt.want)
}
}
}

View File

@@ -0,0 +1,104 @@
package domain
import "testing"
func TestIsValidTier(t *testing.T) {
t.Parallel()
valid := []string{"free", "premium", "pro"}
for _, tier := range valid {
if !IsValidTier(tier) {
t.Errorf("expected %q to be valid", tier)
}
}
invalid := []string{"", "basic", "enterprise", "FREE", "Pro"}
for _, tier := range invalid {
if IsValidTier(tier) {
t.Errorf("expected %q to be invalid", tier)
}
}
}
func TestGetTierLimits_Free(t *testing.T) {
t.Parallel()
limits := GetTierLimits(TierFree)
if limits.MaxAddresses != 1 {
t.Errorf("free MaxAddresses = %d, want 1", limits.MaxAddresses)
}
if limits.MaxAlertTypes != 1 {
t.Errorf("free MaxAlertTypes = %d, want 1", limits.MaxAlertTypes)
}
if len(limits.AllowedChannels) != 1 || limits.AllowedChannels[0] != "email" {
t.Errorf("free AllowedChannels = %v, want [email]", limits.AllowedChannels)
}
if limits.IsUnlimitedAddresses() {
t.Error("free should not have unlimited addresses")
}
if limits.IsUnlimitedAlertTypes() {
t.Error("free should not have unlimited alert types")
}
}
func TestGetTierLimits_Premium(t *testing.T) {
t.Parallel()
limits := GetTierLimits(TierPremium)
if limits.MaxAddresses != 3 {
t.Errorf("premium MaxAddresses = %d, want 3", limits.MaxAddresses)
}
if limits.MaxAlertTypes != 2 {
t.Errorf("premium MaxAlertTypes = %d, want 2", limits.MaxAlertTypes)
}
if !limits.ChannelAllowed("email") {
t.Error("premium should allow email")
}
if !limits.ChannelAllowed("discord") {
t.Error("premium should allow discord")
}
if !limits.ChannelAllowed("telegram") {
t.Error("premium should allow telegram")
}
if limits.ChannelAllowed("slack") {
t.Error("premium should NOT allow slack")
}
}
func TestGetTierLimits_Pro(t *testing.T) {
t.Parallel()
limits := GetTierLimits(TierPro)
if !limits.IsUnlimitedAddresses() {
t.Error("pro should have unlimited addresses")
}
if !limits.IsUnlimitedAlertTypes() {
t.Error("pro should have unlimited alert types")
}
for _, ch := range []string{"email", "discord", "telegram", "slack"} {
if !limits.ChannelAllowed(ch) {
t.Errorf("pro should allow %s", ch)
}
}
}
func TestGetTierLimits_Unknown(t *testing.T) {
t.Parallel()
limits := GetTierLimits(SubscriptionTier("unknown"))
if limits.MaxAddresses != 1 {
t.Errorf("unknown tier should default to free limits, got MaxAddresses=%d", limits.MaxAddresses)
}
}
func TestChannelAllowed_NotInList(t *testing.T) {
t.Parallel()
limits := GetTierLimits(TierFree)
if limits.ChannelAllowed("discord") {
t.Error("free should not allow discord")
}
if limits.ChannelAllowed("nonexistent") {
t.Error("nonexistent channel should not be allowed")
}
}

View File

@@ -384,11 +384,26 @@ func (h *StripeHandler) handleSubscriptionUpdated(r *http.Request, event stripe.
}
status := string(sub.Status)
if err := h.users.UpdateSubscriptionStatus(r.Context(), customerID, status); err != nil {
log.Printf("Failed to update subscription status: %v", err)
// Detect tier from the subscription's current price so that
// upgrades/downgrades via the Stripe portal are reflected.
tier := domain.TierPremium
if sub.Items != nil {
for _, item := range sub.Items.Data {
if item.Price != nil {
if t := h.cfg.TierForPriceID(item.Price.ID); t != "" {
tier = domain.SubscriptionTier(t)
break
}
}
}
}
log.Printf("Subscription %s updated to %s for customer %s", sub.ID, status, customerID)
if err := h.users.ActivateSubscription(r.Context(), customerID, sub.ID, status, tier); err != nil {
log.Printf("Failed to update subscription: %v", err)
}
log.Printf("Subscription %s updated to %s (tier %s) for customer %s", sub.ID, status, tier, customerID)
}
func (h *StripeHandler) handleSubscriptionDeleted(r *http.Request, event stripe.Event) {

View File

@@ -0,0 +1,60 @@
package middleware
import (
"context"
"testing"
)
func TestGetUserID_Empty(t *testing.T) {
t.Parallel()
ctx := context.Background()
if id := GetUserID(ctx); id != "" {
t.Errorf("expected empty user ID, got %q", id)
}
}
func TestGetUserID_Set(t *testing.T) {
t.Parallel()
ctx := context.WithValue(context.Background(), UserIDKey, "abc-123")
if id := GetUserID(ctx); id != "abc-123" {
t.Errorf("expected abc-123, got %q", id)
}
}
func TestGetUserEmail_Empty(t *testing.T) {
t.Parallel()
ctx := context.Background()
if email := GetUserEmail(ctx); email != "" {
t.Errorf("expected empty email, got %q", email)
}
}
func TestGetUserEmail_Set(t *testing.T) {
t.Parallel()
ctx := context.WithValue(context.Background(), UserEmailKey, "test@example.com")
if email := GetUserEmail(ctx); email != "test@example.com" {
t.Errorf("expected test@example.com, got %q", email)
}
}
func TestGetUserTier_Default(t *testing.T) {
t.Parallel()
ctx := context.Background()
if tier := GetUserTier(ctx); tier != "free" {
t.Errorf("expected default tier 'free', got %q", tier)
}
}
func TestGetUserTier_Set(t *testing.T) {
t.Parallel()
ctx := context.WithValue(context.Background(), UserTierKey, "pro")
if tier := GetUserTier(ctx); tier != "pro" {
t.Errorf("expected 'pro', got %q", tier)
}
}

View File

@@ -55,12 +55,14 @@ func (m *AddressModel) ListByUser(ctx context.Context, userID string) ([]domain.
return addresses, rows.Err()
}
// ListAll returns all addresses system-wide (used by the poller).
// ListAll returns addresses for users with an active subscription (used by the poller).
func (m *AddressModel) ListAll(ctx context.Context) ([]domain.Address, error) {
rows, err := m.pool.Query(ctx,
`SELECT id, user_id, address, label, created_at
FROM addresses
ORDER BY created_at DESC`,
`SELECT a.id, a.user_id, a.address, a.label, a.created_at
FROM addresses a
JOIN users u ON u.id = a.user_id
WHERE u.subscription_status IN ('active', 'trialing')
ORDER BY a.created_at DESC`,
)
if err != nil {
return nil, err

View File

@@ -77,10 +77,12 @@ func (m *NotificationConfigModel) Remove(ctx context.Context, userID string) (bo
func (m *NotificationConfigModel) ListEnabled(ctx context.Context) ([]domain.NotificationConfig, error) {
rows, err := m.pool.Query(ctx,
`SELECT user_id, discord_webhook_url, telegram_chat_id, telegram_bot_token,
email, slack_webhook_url
FROM user_notification_configs
WHERE notification_enabled = TRUE`,
`SELECT nc.user_id, nc.discord_webhook_url, nc.telegram_chat_id, nc.telegram_bot_token,
nc.email, nc.slack_webhook_url
FROM user_notification_configs nc
JOIN users u ON u.id = nc.user_id
WHERE nc.notification_enabled = TRUE
AND u.subscription_status IN ('active', 'trialing')`,
)
if err != nil {
return nil, err

View File

@@ -28,6 +28,7 @@ type EvaluatorService struct {
alertRules *models.AlertRuleModel
alertEvents *models.AlertEventModel
addresses *models.AddressModel
users *models.UserModel
notifConfigs *models.NotificationConfigModel
notifSem *semaphore.Weighted
notifWg sync.WaitGroup
@@ -38,6 +39,7 @@ func NewEvaluatorService(
alertRules *models.AlertRuleModel,
alertEvents *models.AlertEventModel,
addresses *models.AddressModel,
users *models.UserModel,
notifConfigs *models.NotificationConfigModel,
) *EvaluatorService {
return &EvaluatorService{
@@ -45,6 +47,7 @@ func NewEvaluatorService(
alertRules: alertRules,
alertEvents: alertEvents,
addresses: addresses,
users: users,
notifConfigs: notifConfigs,
notifSem: semaphore.NewWeighted(maxConcurrentNotifications),
}
@@ -248,14 +251,16 @@ func (s *EvaluatorService) WaitForNotifications() {
s.notifWg.Wait()
}
func (s *EvaluatorService) buildNotifiers(cfg *domain.NotificationConfig) []notifications.Notifier {
func (s *EvaluatorService) buildNotifiers(cfg *domain.NotificationConfig, limits domain.TierLimits) []notifications.Notifier {
var notifiers []notifications.Notifier
if cfg.DiscordWebhookURL != nil && *cfg.DiscordWebhookURL != "" {
if limits.ChannelAllowed("discord") &&
cfg.DiscordWebhookURL != nil && *cfg.DiscordWebhookURL != "" {
notifiers = append(notifiers, &notifications.DiscordNotifier{WebhookURL: *cfg.DiscordWebhookURL})
}
if cfg.TelegramBotToken != nil && *cfg.TelegramBotToken != "" &&
if limits.ChannelAllowed("telegram") &&
cfg.TelegramBotToken != nil && *cfg.TelegramBotToken != "" &&
cfg.TelegramChatID != nil && *cfg.TelegramChatID != "" {
notifiers = append(notifiers, &notifications.TelegramNotifier{
BotToken: *cfg.TelegramBotToken,
@@ -263,7 +268,8 @@ func (s *EvaluatorService) buildNotifiers(cfg *domain.NotificationConfig) []noti
})
}
if cfg.SlackWebhookURL != nil && *cfg.SlackWebhookURL != "" {
if limits.ChannelAllowed("slack") &&
cfg.SlackWebhookURL != nil && *cfg.SlackWebhookURL != "" {
notifiers = append(notifiers, &notifications.SlackNotifier{WebhookURL: *cfg.SlackWebhookURL})
}
@@ -313,6 +319,13 @@ func (s *EvaluatorService) sendNotification(ctx context.Context, userID, message
return
}
user, err := s.users.GetByID(ctx, userID)
if err != nil || user == nil {
log.Printf("Failed to load user %s for tier check in notification: %v", userID, err)
return
}
limits := domain.GetTierLimits(user.SubscriptionTier)
meta := notifications.AlertMetadata{
TxHash: obs.Hash,
AddressLabel: addressLabel,
@@ -320,7 +333,7 @@ func (s *EvaluatorService) sendNotification(ctx context.Context, userID, message
Address: address,
}
for _, n := range s.buildNotifiers(notifConfig) {
for _, n := range s.buildNotifiers(notifConfig, limits) {
if err := sendWithRetry(ctx, n, message, meta); err != nil {
log.Printf("Notification channel failed for user %s after retries: %v", userID, err)
} else {

Binary file not shown.