notifications

package
v0.0.0-...-6deb405 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Apr 12, 2026 License: MIT Imports: 22 Imported by: 0

README

Push Notification Infrastructure

This package provides a flexible push notification infrastructure for CryptoFunk with pluggable backends.

Features

  • Pluggable Backends: Start with Firebase Cloud Messaging (FCM), easily extend to APNs, web push, etc.
  • User Preferences: Fine-grained control over notification types
  • Device Management: Track and manage multiple devices per user
  • Notification Logging: Audit trail of all sent notifications
  • Mock Support: Development-friendly mock backend when FCM credentials aren't configured

Architecture

NotificationService (service.go)
    ├─ Backend Interface (pluggable)
    │   └─ FCMBackend (fcm.go)
    ├─ Database Layer (user_devices, notification_preferences, notification_log)
    └─ Helper Methods (helpers.go)

Database Schema

user_devices

Stores device tokens for push notifications.

Column Type Description
id UUID Primary key
user_id UUID Foreign key to users
device_token TEXT FCM/APNs device token
platform VARCHAR(20) ios, android, or web
enabled BOOLEAN Whether device is active
created_at TIMESTAMP Registration time
last_used_at TIMESTAMP Last notification sent
notification_preferences

User preferences for different notification types.

Column Type Description
user_id UUID Primary key, foreign key to users
trade_executions BOOLEAN Trade execution notifications
pnl_alerts BOOLEAN P&L change alerts
circuit_breaker BOOLEAN Circuit breaker triggers
consensus_failures BOOLEAN Agent consensus failures
notification_log

Audit log of all sent notifications (TimescaleDB hypertable).

Column Type Description
id UUID Primary key
user_id UUID Recipient user
device_token TEXT Target device
notification_type VARCHAR(50) Type of notification
title TEXT Notification title
body TEXT Notification body
data JSONB Additional payload
status VARCHAR(20) pending, sent, or failed
error_message TEXT Error if failed
sent_at TIMESTAMP When sent

Usage

Initialize the Service
package main

import (
    "context"
    "github.com/ajitpratap0/cryptofunk/internal/notifications"
    "github.com/jackc/pgx/v5/pgxpool"
)

func main() {
    ctx := context.Background()

    // Connect to database
    db, _ := pgxpool.New(ctx, "postgres://...")
    defer db.Close()

    // Initialize FCM backend (uses mock if no credentials)
    backend, _ := notifications.NewFCMBackend(ctx, "/path/to/fcm-credentials.json")

    // Create notification service
    service := notifications.NewService(db, backend)
    defer service.Close()
}
Register a Device
err := service.RegisterDevice(ctx, userID, deviceToken, notifications.PlatformIOS)
if err != nil {
    log.Error().Err(err).Msg("Failed to register device")
}
Send Notifications
Using the Service Directly
notification := notifications.Notification{
    Type:     notifications.NotificationTypeTradeExecution,
    Title:    "Trade Executed",
    Body:     "Your BTC/USDT order has been filled",
    Data: map[string]string{
        "order_id": "12345",
        "symbol":   "BTC/USDT",
    },
    Priority: "high",
}

err := service.SendToUser(ctx, userID, notification)
Using Helper Methods
helper := notifications.NewHelper(service)

// Trade execution
helper.SendTradeExecution(ctx, userID, "order-123", "BTC/USDT", "BUY", 0.5, 50000.0)

// P&L alert (when change exceeds 5%)
helper.SendPnLAlert(ctx, userID, "session-456", 7.5, 1500.0)

// Circuit breaker triggered
helper.SendCircuitBreakerAlert(ctx, userID, "max_drawdown_exceeded", 10.0)

// Consensus failure
helper.SendConsensusFailure(ctx, userID, "ETH/USDT", "insufficient_confidence", 5)
Manage User Preferences
// Update preferences
prefs := notifications.Preferences{
    TradeExecutions:   true,
    PnLAlerts:         true,
    CircuitBreaker:    true,
    ConsensusFailures: false, // User opts out
}
err := service.UpdatePreferences(ctx, userID, prefs)

// Get current preferences
prefs, err := service.GetPreferences(ctx, userID)
Integration Points
1. Trade Execution (in order execution code)
// In internal/exchange/executor.go or similar
func (e *Executor) notifyTradeExecution(order *Order) {
    helper := notifications.NewHelper(e.notificationService)
    helper.SendTradeExecution(
        context.Background(),
        order.UserID,
        order.ID,
        order.Symbol,
        order.Side,
        order.Quantity,
        order.FilledPrice,
    )
}
2. P&L Monitoring (in position management)
// In internal/positions/manager.go or similar
func (m *Manager) checkPnLThreshold(userID, sessionID string, oldPnL, newPnL float64) {
    helper := notifications.NewHelper(m.notificationService)
    helper.CheckPnLThresholdAndNotify(
        context.Background(),
        userID,
        sessionID,
        oldPnL,
        newPnL,
        5.0, // 5% threshold
    )
}
3. Circuit Breaker (in risk management)
// In internal/risk/circuit_breaker.go or similar
func (cb *CircuitBreaker) triggerBreaker(reason string) {
    helper := notifications.NewHelper(cb.notificationService)

    // Notify all active users
    users := cb.getActiveUsers()
    for _, userID := range users {
        helper.SendCircuitBreakerAlert(
            context.Background(),
            userID,
            reason,
            cb.threshold,
        )
    }
}
4. Consensus Failures (in orchestrator)
// In internal/orchestrator/consensus.go or similar
func (o *Orchestrator) handleConsensusFailure(symbol, reason string, agentCount int) {
    helper := notifications.NewHelper(o.notificationService)

    // Notify subscribed users
    users := o.getSubscribedUsers(symbol)
    for _, userID := range users {
        helper.SendConsensusFailure(
            context.Background(),
            userID,
            symbol,
            reason,
            agentCount,
        )
    }
}

Configuration

FCM Setup
  1. Create a Firebase project at https://console.firebase.google.com
  2. Generate a service account key:
    • Project Settings > Service Accounts > Generate New Private Key
  3. Save the JSON file securely
  4. Set the path in your configuration or environment:
# configs/config.yaml
notifications:
  fcm_credentials_path: "/path/to/firebase-credentials.json"
Environment Variables
export FCM_CREDENTIALS_PATH="/path/to/firebase-credentials.json"
Mock Mode (Development)

If no FCM credentials are configured, the service automatically uses a mock backend that logs notifications to stderr instead of sending them. This is perfect for development and testing.

// This will use mock mode if file doesn't exist
backend, _ := notifications.NewFCMBackend(ctx, "")
// Logs: "Mock FCM notification (not actually sent)"

Notification Types

Trade Execution
  • When: Order is filled
  • Priority: High
  • Default: Enabled
P&L Alerts
  • When: P&L changes by ±5% or more
  • Priority: High
  • Default: Enabled
Circuit Breaker
  • When: Trading is halted due to risk limits
  • Priority: High
  • Default: Enabled
Consensus Failures
  • When: Agents fail to reach consensus on a trade
  • Priority: Normal
  • Default: Enabled

Testing

# Run all notification tests
go test -v ./internal/notifications/...

# Test with coverage
go test -v -cover ./internal/notifications/...

# Test specific function
go test -v -run TestFCMBackend ./internal/notifications/...

Security Considerations

  1. Device Token Security: Device tokens are sensitive and should be transmitted over HTTPS only
  2. FCM Credentials: Store Firebase credentials securely, never commit to git
  3. User Privacy: Respect user preferences, provide easy opt-out
  4. Rate Limiting: Consider implementing rate limits to prevent notification spam
  5. Token Masking: Device tokens are automatically masked in logs (e.g., "abcd...5678")

Performance

  • Database Indexes: All query patterns are indexed for fast lookups
  • TimescaleDB: notification_log uses TimescaleDB for efficient time-series queries
  • Batch Sending: Use FCMBackend.SendMulticast for sending to multiple devices
  • Connection Pool: Uses pgxpool for efficient database connections

Extending with New Backends

To add a new notification backend (e.g., APNs, Web Push):

type MyBackend struct {
    // backend-specific fields
}

func (b *MyBackend) Send(ctx context.Context, deviceToken string, notification Notification) error {
    // implementation
}

func (b *MyBackend) Name() string {
    return "my_backend"
}

func (b *MyBackend) Close() error {
    return nil
}

// Use it
backend := &MyBackend{}
service := notifications.NewService(db, backend)

Migration

Run the migration to create the required tables:

task db-migrate
# Or manually:
psql -U cryptofunk -d cryptofunk -f migrations/011_user_devices.sql

Troubleshooting

"Failed to send FCM message"
  • Check FCM credentials path is correct
  • Verify device token is valid
  • Ensure Firebase project has Cloud Messaging enabled
"Device token not found"
  • Device must be registered first with RegisterDevice
  • Check device is not disabled
"Notification type disabled for user"
  • User has opted out of this notification type
  • Check user preferences with GetPreferences

Future Enhancements

  • Add APNs backend for iOS native support
  • Add web push backend for browser notifications
  • Implement notification batching for efficiency
  • Add retry logic with exponential backoff
  • Support notification templates
  • Add notification scheduling
  • Implement notification channels/topics
  • Add A/B testing support for notification content

Documentation

Overview

Example (BasicUsage)

Example_basicUsage demonstrates basic notification service usage 

package main

import (
	"context"
	"fmt"

	"github.com/jackc/pgx/v5/pgxpool"

	"github.com/ajitpratap0/cryptofunk/internal/notifications"
)

func main() {
	ctx := context.Background()

	// In production, use real database connection
	// db, _ := pgxpool.New(ctx, "postgres://...")
	var db *pgxpool.Pool // nil for example

	// Initialize FCM backend (mock mode for development)
	backend, _ := notifications.NewFCMBackend(ctx, "")

	// Create notification service
	service := notifications.NewService(db, backend)
	defer service.Close()

	// Create notification
	notification := notifications.Notification{
		Type:     notifications.NotificationTypeTradeExecution,
		Title:    "Trade Executed",
		Body:     "Your BTC/USDT order has been filled",
		Data:     notifications.TradeNotificationData("order-123", "BTC/USDT", "BUY", 0.5, 50000.0),
		Priority: "high",
	}

	fmt.Println("Notification type:", notification.Type)
	fmt.Println("Title:", notification.Title)
}

Output:
Notification type: trade_execution
Title: Trade Executed
Example (DeviceRegistration)

Example_deviceRegistration demonstrates device registration flow 

package main

import (
	"context"
	"fmt"

	"github.com/ajitpratap0/cryptofunk/internal/notifications"
)

func main() {
	ctx := context.Background()

	// Mock setup
	backend, _ := notifications.NewFCMBackend(ctx, "")
	service := notifications.NewService(nil, backend)

	// Simulate device registration (would require database in production)
	userID := "user-123"
	deviceToken := "fcm-device-token-here"
	platform := notifications.PlatformIOS

	fmt.Printf("Registering device for user %s\n", userID)
	fmt.Printf("Platform: %s\n", platform)
	fmt.Printf("Token length: %d\n", len(deviceToken))

	// In production:
	// err := service.RegisterDevice(ctx, userID, deviceToken, platform)
	// if err != nil {
	//     log.Error().Err(err).Msg("Failed to register device")
	// }

	_ = service

}

Output:
Registering device for user user-123
Platform: ios
Token length: 21
Example (HelperMethods)

Example_helperMethods demonstrates using notification helper methods 

package main

import (
	"context"
	"fmt"

	"github.com/ajitpratap0/cryptofunk/internal/notifications"
)

func main() {
	ctx := context.Background()

	// Setup (mock for demonstration)
	backend, _ := notifications.NewFCMBackend(ctx, "")
	service := notifications.NewService(nil, backend)
	helper := notifications.NewHelper(service)

	// These would normally send notifications to users
	_ = helper
	_ = ctx

	// Example notification data
	tradeData := notifications.TradeNotificationData("order-123", "BTC/USDT", "BUY", 0.5, 50000.0)
	fmt.Println("Order ID:", tradeData["order_id"])
	fmt.Println("Symbol:", tradeData["symbol"])
	fmt.Println("Price:", tradeData["price"])

}

Output:
Order ID: order-123
Symbol: BTC/USDT
Price: 50000.00
Example (IntegrationWithOrchestrator)

Example_integrationWithOrchestrator shows how to integrate with the orchestrator 

package main

import (
	"context"
	"fmt"
	"log"

	"github.com/jackc/pgx/v5/pgxpool"

	"github.com/ajitpratap0/cryptofunk/internal/notifications"
)

func main() {
	// This example shows how to integrate notifications in the orchestrator

	// 1. Initialize the notification service in main()
	ctx := context.Background()
	backend, err := notifications.NewFCMBackend(ctx, "/path/to/fcm-credentials.json")
	if err != nil {
		log.Fatal(err)
	}

	// In production, connect to real database
	// db, _ := pgxpool.New(ctx, "postgres://...")
	var db *pgxpool.Pool

	service := notifications.NewService(db, backend)
	helper := notifications.NewHelper(service)

	// 2. Inject into orchestrator/components
	_ = helper

	// 3. Send notifications on events
	// Example: After trade execution
	// helper.SendTradeExecution(ctx, userID, orderID, symbol, side, quantity, price)

	// Example: On P&L threshold breach
	// helper.SendPnLAlert(ctx, userID, sessionID, pnlPercent, pnlAmount)

	// Example: Circuit breaker triggered
	// helper.SendCircuitBreakerAlert(ctx, userID, reason, threshold)

	// Example: Consensus failure
	// helper.SendConsensusFailure(ctx, userID, symbol, reason, agentCount)

	fmt.Println("Notification service initialized")
}

Output:
Notification service initialized
Example (Preferences)

Example_preferences demonstrates notification preferences 

package main

import (
	"fmt"

	"github.com/ajitpratap0/cryptofunk/internal/notifications"
)

func main() {
	// Default preferences (all enabled)
	prefs := notifications.DefaultPreferences()
	fmt.Println("Trade executions:", prefs.TradeExecutions)
	fmt.Println("P&L alerts:", prefs.PnLAlerts)
	fmt.Println("Circuit breaker:", prefs.CircuitBreaker)
	fmt.Println("Consensus failures:", prefs.ConsensusFailures)

	// Check if specific type is enabled
	enabled := prefs.IsEnabled(notifications.NotificationTypeTradeExecution)
	fmt.Println("Trade notifications enabled:", enabled)

}

Output:
Trade executions: true
P&L alerts: true
Circuit breaker: true
Consensus failures: true
Trade notifications enabled: true

Index

Examples

Constants

This section is empty.

Variables

This section is empty.

Functions

func CircuitBreakerNotificationData 

func CircuitBreakerNotificationData(reason string, threshold float64) map[string]string

CircuitBreakerNotificationData creates data payload for circuit breaker alerts

func ConsensusFailureNotificationData 

func ConsensusFailureNotificationData(symbol, reason string, agentCount int) map[string]string

ConsensusFailureNotificationData creates data payload for consensus failure alerts

func DailySummaryNotificationData 

func DailySummaryNotificationData(
	date string,
	totalTrades, winningTrades, losingTrades int,
	totalPnL, totalPnLPercent, winRate float64,
) map[string]string

DailySummaryNotificationData creates data payload for daily summary notifications

func NewEmailChannel 

func NewEmailChannel(cfg EmailChannelConfig) *emailChannel

NewEmailChannel creates a new email Channel implementation

func PnLNotificationData 

func PnLNotificationData(sessionID string, pnlPercent, pnlAmount float64) map[string]string

PnLNotificationData creates data payload for P&L alerts

func PositionClosedNotificationData 

func PositionClosedNotificationData(positionID, symbol, side string, quantity, entryPrice, exitPrice, pnl, pnlPercent float64) map[string]string

PositionClosedNotificationData creates data payload for position closed notifications

func SafetyGuardNotificationData 

func SafetyGuardNotificationData(guardType, reason string, currentValue, threshold float64) map[string]string

SafetyGuardNotificationData creates data payload for safety guard triggered alerts

func SystemErrorNotificationData 

func SystemErrorNotificationData(component, errorType, errorMsg string) map[string]string

SystemErrorNotificationData creates data payload for system error notifications

func TradeNotificationData 

func TradeNotificationData(orderID, symbol, side string, quantity, price float64) map[string]string

TradeNotificationData creates data payload for trade execution notifications

func ValidateToken 

func ValidateToken(token string) bool

ValidateToken checks if a device token is valid for FCM This is a simple validation - actual validation happens when sending

Types

type Backend 

type Backend interface {
	// Send sends a notification to a device
	Send(ctx context.Context, deviceToken string, notification Notification) error

	// Name returns the backend name
	Name() string

	// Close closes the backend connection
	Close() error
}

Backend defines the interface for notification backends (FCM, APNs, etc.)

type Channel 

type Channel interface {
	// Name returns the channel identifier
	Name() string
	// Send delivers a notification event
	Send(ctx context.Context, event Event) error
	// Close cleans up resources
	Close() error
}

Channel is the interface for notification delivery channels

func NewSlackChannel 

func NewSlackChannel(cfg SlackChannelConfig) Channel

NewSlackChannel creates a new Slack Channel implementation

type ChannelType 

type ChannelType string

Channel represents a notification channel 

const (
	ChannelEmail ChannelType = "email"
	ChannelSlack ChannelType = "slack"
)

type Device 

type Device struct {
	ID          string    `json:"id"`
	UserID      string    `json:"user_id"`
	DeviceToken string    `json:"device_token"`
	Platform    Platform  `json:"platform"`
	Enabled     bool      `json:"enabled"`
	CreatedAt   time.Time `json:"created_at"`
	LastUsedAt  time.Time `json:"last_used_at"`
}

Device represents a user device for push notifications

type EmailBackend 

type EmailBackend struct {
	// contains filtered or unexported fields
}

EmailBackend implements the Backend interface for sending email notifications

func NewEmailBackend 

func NewEmailBackend(config EmailConfig, defaultRecipients []string) (*EmailBackend, error)

NewEmailBackend creates a new email notification backend If config is incomplete (missing host/from_address), creates a mock backend

func (*EmailBackend) AddRecipient 

func (e *EmailBackend) AddRecipient(email string)

AddRecipient adds a default recipient

func (*EmailBackend) Close 

func (e *EmailBackend) Close() error

Close closes the email backend

func (*EmailBackend) GetRecipients 

func (e *EmailBackend) GetRecipients() []string

GetRecipients returns the list of default recipients

func (*EmailBackend) IsMock 

func (e *EmailBackend) IsMock() bool

IsMock returns true if this is a mock backend

func (*EmailBackend) Name 

func (e *EmailBackend) Name() string

Name returns the backend name

func (*EmailBackend) RemoveRecipient 

func (e *EmailBackend) RemoveRecipient(email string)

RemoveRecipient removes a default recipient

func (*EmailBackend) Send 

func (e *EmailBackend) Send(ctx context.Context, deviceToken string, notification Notification) error

Send sends a notification via email

func (*EmailBackend) SetRecipients 

func (e *EmailBackend) SetRecipients(recipients []string)

SetRecipients sets the list of default recipients

type EmailChannelConfig 

type EmailChannelConfig struct {
	SMTPHost  string
	SMTPPort  string
	From      string
	To        string
	Password  string
	BatchMode bool
	SendFunc  func(addr string, a smtp.Auth, from string, to []string, msg []byte) error
}

EmailChannelConfig holds config for an email notification channel

type EmailConfig 

type EmailConfig struct {
	// SMTP server settings
	Host     string `mapstructure:"host"`
	Port     int    `mapstructure:"port"`
	Username string `mapstructure:"username"`
	Password string `mapstructure:"password"`

	// Sender information
	FromAddress string `mapstructure:"from_address"`
	FromName    string `mapstructure:"from_name"`

	// Security settings
	UseTLS       bool `mapstructure:"use_tls"`       // Use TLS from start (port 465)
	UseStartTLS  bool `mapstructure:"use_starttls"`  // Use STARTTLS upgrade (port 587)
	SkipVerify   bool `mapstructure:"skip_verify"`   // Skip TLS certificate verification (dev only)
	AuthRequired bool `mapstructure:"auth_required"` // Whether SMTP auth is required

	// Rate limiting
	RateLimitPerMinute int `mapstructure:"rate_limit_per_minute"` // Max emails per minute (0 = unlimited)
	RetryAttempts      int `mapstructure:"retry_attempts"`        // Number of retry attempts on failure
	RetryDelaySeconds  int `mapstructure:"retry_delay_seconds"`   // Delay between retries
}

EmailConfig holds SMTP configuration for email notifications

func DefaultEmailConfig 

func DefaultEmailConfig() EmailConfig

DefaultEmailConfig returns default email configuration

type Event 

type Event struct {
	Type      EventType         `json:"type"`
	Priority  Priority          `json:"priority"`
	Title     string            `json:"title"`
	Message   string            `json:"message"`
	Fields    map[string]string `json:"fields,omitempty"`
	Timestamp time.Time         `json:"timestamp"`
}

Event represents a notification event to be dispatched

func DailySummaryEvent 

func DailySummaryEvent(totalTrades int, pnl, winRate float64, bestTrade, worstTrade string) Event

DailySummaryEvent creates a daily summary notification event

func ErrorAlertEvent 

func ErrorAlertEvent(source, message string) Event

ErrorAlertEvent creates an error alert notification event

func PositionClosedEvent 

func PositionClosedEvent(symbol string, pnl float64, holdDuration time.Duration, reason string) Event

PositionClosedEvent creates a position closed notification event

func SafetyAlertEvent 

func SafetyAlertEvent(alertType, details string) Event

SafetyAlertEvent creates a safety alert notification event

func TradeExecutedEvent 

func TradeExecutedEvent(symbol, side, agent string, price, size float64) Event

TradeExecutedEvent creates a trade executed notification event

type EventConfig 

type EventConfig struct {
	Channels []ChannelType `mapstructure:"channels" json:"channels"`
	Priority string        `mapstructure:"priority" json:"priority"` // "high" or "normal"
	Enabled  bool          `mapstructure:"enabled" json:"enabled"`
}

EventConfig maps event types to channels and priority settings

type EventEmitter 

type EventEmitter interface {
	SetNotificationCallback(callback NotificationCallback)
}

EventEmitter is an interface for components that can emit notification events

type EventType 

type EventType string

EventType represents different notification event types 

const (
	EventTradeExecuted  EventType = "trade_executed"
	EventPositionClosed EventType = "position_closed"
	EventErrorAlert     EventType = "error_alert"
	EventSafetyAlert    EventType = "safety_alert"
	EventDailySummary   EventType = "daily_summary"
)

type FCMBackend 

type FCMBackend struct {
	// contains filtered or unexported fields
}

FCMBackend implements the Backend interface using Firebase Cloud Messaging

func NewFCMBackend 

func NewFCMBackend(ctx context.Context, credentialsPath string) (*FCMBackend, error)

NewFCMBackend creates a new FCM backend If credentialsPath is empty or file doesn't exist, creates a mock backend

func (*FCMBackend) Close 

func (f *FCMBackend) Close() error

Close closes the FCM backend

func (*FCMBackend) IsMock 

func (f *FCMBackend) IsMock() bool

IsMock returns true if this is a mock backend

func (*FCMBackend) Name 

func (f *FCMBackend) Name() string

Name returns the backend name

func (*FCMBackend) Send 

func (f *FCMBackend) Send(ctx context.Context, deviceToken string, notification Notification) error

Send sends a notification via FCM

func (*FCMBackend) SendMulticast 

func (f *FCMBackend) SendMulticast(ctx context.Context, deviceTokens []string, notification Notification) (*messaging.BatchResponse, error)

SendMulticast sends a notification to multiple devices

type Integration 

type Integration struct {
	// contains filtered or unexported fields
}

Integration provides notification integration for various components It acts as a bridge between system events and the notification manager

func NewIntegration 

func NewIntegration(manager *Manager) *Integration

NewIntegration creates a new notification integration

func (*Integration) CreateSafetyGuardCallback 

func (i *Integration) CreateSafetyGuardCallback() func(event interface{})

CreateSafetyGuardCallback creates a callback function for safety guard events This can be passed to SafetyGuard.SetMetricsCallback to emit notifications

func (*Integration) EmitCircuitBreakerTriggered 

func (i *Integration) EmitCircuitBreakerTriggered(ctx context.Context, reason string, threshold float64)

EmitCircuitBreakerTriggered emits a circuit breaker triggered notification

func (*Integration) EmitConsensusFailure 

func (i *Integration) EmitConsensusFailure(ctx context.Context, symbol, reason string, agentCount int)

EmitConsensusFailure emits a consensus failure notification

func (*Integration) EmitDailySummary 

func (i *Integration) EmitDailySummary(ctx context.Context, date string, totalTrades, winningTrades, losingTrades int, totalPnL, totalPnLPercent, winRate float64)

EmitDailySummary emits a daily summary notification

func (*Integration) EmitPnLAlert 

func (i *Integration) EmitPnLAlert(ctx context.Context, sessionID string, pnlPercent, pnlAmount float64)

EmitPnLAlert emits a P&L alert notification

func (*Integration) EmitPositionClosed 

func (i *Integration) EmitPositionClosed(ctx context.Context, positionID, symbol, side string, quantity, entryPrice, exitPrice, pnl, pnlPercent float64)

EmitPositionClosed emits a position closed notification

func (*Integration) EmitSafetyGuardTriggered 

func (i *Integration) EmitSafetyGuardTriggered(ctx context.Context, guardType, reason string, currentValue, threshold float64)

EmitSafetyGuardTriggered emits a safety guard triggered notification

func (*Integration) EmitSystemError 

func (i *Integration) EmitSystemError(ctx context.Context, component, errorType, errorMsg string)

EmitSystemError emits a system error notification

func (*Integration) EmitTradeExecuted 

func (i *Integration) EmitTradeExecuted(ctx context.Context, orderID, symbol, side string, quantity, price float64)

EmitTradeExecuted emits a trade execution notification

func (*Integration) GetStats 

func (i *Integration) GetStats() map[string]interface{}

GetStats returns integration statistics

func (*Integration) IsEnabled 

func (i *Integration) IsEnabled() bool

IsEnabled returns whether the integration is enabled

func (*Integration) SetCooldown 

func (i *Integration) SetCooldown(notifType NotificationType, duration time.Duration)

SetCooldown sets the cooldown duration for a specific notification type

func (*Integration) SetEnabled 

func (i *Integration) SetEnabled(enabled bool)

SetEnabled enables or disables the integration

type Manager 

type Manager struct {
	// contains filtered or unexported fields
}

Manager routes notifications to appropriate channels based on event type

func NewManager 

func NewManager(config ManagerConfig, emailBackend *EmailBackend, slackBackend *SlackBackend) *Manager

NewManager creates a new notification manager

func NewManagerFromConfig 

func NewManagerFromConfig(cfg *config.Config) (*Manager, error)

NewManagerFromConfig creates a NotificationManager from the application config

func (*Manager) Close 

func (m *Manager) Close() error

Close closes all backends

func (*Manager) GetChannelsForEvent 

func (m *Manager) GetChannelsForEvent(notifType NotificationType) []ChannelType

GetChannelsForEvent returns the channels configured for a specific event type

func (*Manager) GetStats 

func (m *Manager) GetStats() map[string]interface{}

GetStats returns notification statistics

func (*Manager) IsEnabled 

func (m *Manager) IsEnabled() bool

IsEnabled returns whether notifications are enabled

func (*Manager) IsEventEnabled 

func (m *Manager) IsEventEnabled(notifType NotificationType) bool

IsEventEnabled returns whether a specific event type is enabled

func (*Manager) Send 

func (m *Manager) Send(ctx context.Context, notification Notification) error

Send routes a notification to the appropriate channels based on configuration

func (*Manager) SendCircuitBreakerTriggered 

func (m *Manager) SendCircuitBreakerTriggered(ctx context.Context, reason string, threshold float64) error

SendCircuitBreakerTriggered sends a circuit breaker triggered notification

func (*Manager) SendConsensusFailure 

func (m *Manager) SendConsensusFailure(ctx context.Context, symbol, reason string, agentCount int) error

SendConsensusFailure sends a consensus failure notification

func (*Manager) SendDailySummary 

func (m *Manager) SendDailySummary(ctx context.Context, date string, totalTrades, winningTrades, losingTrades int, totalPnL, totalPnLPercent, winRate float64) error

SendDailySummary sends a daily summary notification

func (*Manager) SendPnLAlert 

func (m *Manager) SendPnLAlert(ctx context.Context, sessionID string, pnlPercent, pnlAmount float64) error

SendPnLAlert sends a P&L alert notification

func (*Manager) SendPositionClosed 

func (m *Manager) SendPositionClosed(ctx context.Context, positionID, symbol, side string, quantity, entryPrice, exitPrice, pnl, pnlPercent float64) error

SendPositionClosed sends a position closed notification

func (*Manager) SendSafetyGuardTriggered 

func (m *Manager) SendSafetyGuardTriggered(ctx context.Context, guardType, reason string, currentValue, threshold float64) error

SendSafetyGuardTriggered sends a safety guard triggered notification

func (*Manager) SendSystemError 

func (m *Manager) SendSystemError(ctx context.Context, component, errorType, errorMsg string) error

SendSystemError sends a system error notification

func (*Manager) SendTradeExecuted 

func (m *Manager) SendTradeExecuted(ctx context.Context, orderID, symbol, side string, quantity, price float64) error

SendTradeExecuted sends a trade execution notification

type ManagerConfig 

type ManagerConfig struct {
	Enabled bool `mapstructure:"enabled" json:"enabled"`

	// Event routing configuration
	Events map[NotificationType]EventConfig `mapstructure:"events" json:"events"`

	// Default channels if event is not explicitly configured
	DefaultChannels []ChannelType `mapstructure:"default_channels" json:"default_channels"`
}

ManagerConfig holds configuration for the notification manager

func DefaultManagerConfig 

func DefaultManagerConfig() ManagerConfig

DefaultManagerConfig returns sensible defaults for notification routing

type Notification 

type Notification struct {
	Type     NotificationType  `json:"type"`
	Title    string            `json:"title"`
	Body     string            `json:"body"`
	Data     map[string]string `json:"data,omitempty"`
	Priority string            `json:"priority,omitempty"` // "high" or "normal"
}

Notification represents a push notification to be sent

type NotificationCallback 

type NotificationCallback func(ctx context.Context, notification Notification)

NotificationCallback is a function type for notification callbacks

type NotificationHelper 

type NotificationHelper struct {
	// contains filtered or unexported fields
}

NotificationHelper provides convenient methods for sending common notifications

func NewHelper 

func NewHelper(service Service) *NotificationHelper

NewHelper creates a new notification helper

func (*NotificationHelper) BulkSendDailySummary 

func (h *NotificationHelper) BulkSendDailySummary(ctx context.Context, userIDs []string, date string, totalTrades, winningTrades, losingTrades int, totalPnL, totalPnLPercent, winRate float64) error

BulkSendDailySummary sends daily summary notifications to multiple users

func (*NotificationHelper) BulkSendPositionClosed 

func (h *NotificationHelper) BulkSendPositionClosed(ctx context.Context, userIDs []string, positionID, symbol, side string, quantity, entryPrice, exitPrice, pnl, pnlPercent float64) error

BulkSendPositionClosed sends position closed notifications to multiple users

func (*NotificationHelper) BulkSendSafetyGuard 

func (h *NotificationHelper) BulkSendSafetyGuard(ctx context.Context, userIDs []string, guardType, reason string, currentValue, threshold float64) error

BulkSendSafetyGuard sends safety guard notifications to multiple users

func (*NotificationHelper) BulkSendSystemError 

func (h *NotificationHelper) BulkSendSystemError(ctx context.Context, userIDs []string, component, errorType, errorMsg string) error

BulkSendSystemError sends system error notifications to multiple users (typically admins)

func (*NotificationHelper) BulkSendTradeExecution 

func (h *NotificationHelper) BulkSendTradeExecution(ctx context.Context, userIDs []string, orderID, symbol, side string, quantity, price float64) error

BulkSendTradeExecution sends trade execution notifications to multiple users

func (*NotificationHelper) CheckPnLThresholdAndNotify 

func (h *NotificationHelper) CheckPnLThresholdAndNotify(ctx context.Context, userID, sessionID string, previousPnL, currentPnL float64, threshold float64) error

CheckPnLThresholdAndNotify checks if P&L change exceeds threshold and sends notification

func (*NotificationHelper) SendCircuitBreakerAlert 

func (h *NotificationHelper) SendCircuitBreakerAlert(ctx context.Context, userID, reason string, threshold float64) error

SendCircuitBreakerAlert sends a circuit breaker notification

func (*NotificationHelper) SendConsensusFailure 

func (h *NotificationHelper) SendConsensusFailure(ctx context.Context, userID, symbol, reason string, agentCount int) error

SendConsensusFailure sends a consensus failure notification

func (*NotificationHelper) SendDailySummary 

func (h *NotificationHelper) SendDailySummary(ctx context.Context, userID, date string, totalTrades, winningTrades, losingTrades int, totalPnL, totalPnLPercent, winRate float64) error

SendDailySummary sends a daily summary notification

func (*NotificationHelper) SendPnLAlert 

func (h *NotificationHelper) SendPnLAlert(ctx context.Context, userID, sessionID string, pnlPercent, pnlAmount float64) error

SendPnLAlert sends a P&L alert notification

func (*NotificationHelper) SendPositionClosed 

func (h *NotificationHelper) SendPositionClosed(ctx context.Context, userID, positionID, symbol, side string, quantity, entryPrice, exitPrice, pnl, pnlPercent float64) error

SendPositionClosed sends a position closed notification

func (*NotificationHelper) SendSafetyGuardTriggered 

func (h *NotificationHelper) SendSafetyGuardTriggered(ctx context.Context, userID, guardType, reason string, currentValue, threshold float64) error

SendSafetyGuardTriggered sends a safety guard triggered notification

func (*NotificationHelper) SendSystemError 

func (h *NotificationHelper) SendSystemError(ctx context.Context, userID, component, errorType, errorMsg string) error

SendSystemError sends a system error notification

func (*NotificationHelper) SendTradeExecution 

func (h *NotificationHelper) SendTradeExecution(ctx context.Context, userID, orderID, symbol, side string, quantity, price float64) error

SendTradeExecution sends a trade execution notification

type NotificationLog 

type NotificationLog struct {
	ID               string            `json:"id"`
	UserID           string            `json:"user_id"`
	DeviceToken      string            `json:"device_token,omitempty"`
	NotificationType NotificationType  `json:"notification_type"`
	Title            string            `json:"title"`
	Body             string            `json:"body"`
	Data             map[string]string `json:"data,omitempty"`
	Status           string            `json:"status"` // pending, sent, failed
	ErrorMessage     string            `json:"error_message,omitempty"`
	SentAt           time.Time         `json:"sent_at"`
}

NotificationLog represents a logged notification

type NotificationService 

type NotificationService struct {
	// contains filtered or unexported fields
}

NotificationService implements the Service interface

func NewService 

func NewService(db *pgxpool.Pool, backend Backend) *NotificationService

NewService creates a new notification service

func (*NotificationService) Close 

func (s *NotificationService) Close() error

Close closes the notification service

func (*NotificationService) GetPreferences 

func (s *NotificationService) GetPreferences(ctx context.Context, userID string) (Preferences, error)

GetPreferences returns user notification preferences

func (*NotificationService) GetUserDevices 

func (s *NotificationService) GetUserDevices(ctx context.Context, userID string) ([]Device, error)

GetUserDevices returns all enabled devices for a user

func (*NotificationService) RegisterDevice 

func (s *NotificationService) RegisterDevice(ctx context.Context, userID, deviceToken string, platform Platform) error

RegisterDevice registers a new device for push notifications

func (*NotificationService) SendToDevice 

func (s *NotificationService) SendToDevice(ctx context.Context, deviceToken string, notification Notification) error

SendToDevice sends a notification to a specific device

func (*NotificationService) SendToUser 

func (s *NotificationService) SendToUser(ctx context.Context, userID string, notification Notification) error

SendToUser sends a notification to all enabled devices for a user

func (*NotificationService) UnregisterDevice 

func (s *NotificationService) UnregisterDevice(ctx context.Context, deviceToken string) error

UnregisterDevice removes a device token

func (*NotificationService) UpdateDeviceLastUsed 

func (s *NotificationService) UpdateDeviceLastUsed(ctx context.Context, deviceToken string) error

UpdateDeviceLastUsed updates the last used timestamp for a device

func (*NotificationService) UpdatePreferences 

func (s *NotificationService) UpdatePreferences(ctx context.Context, userID string, prefs Preferences) error

UpdatePreferences updates user notification preferences

type NotificationStatus 

type NotificationStatus string

NotificationStatus represents the status of a sent notification 

const (
	NotificationStatusPending NotificationStatus = "pending"
	NotificationStatusSent    NotificationStatus = "sent"
	NotificationStatusFailed  NotificationStatus = "failed"
)

type NotificationType 

type NotificationType string

NotificationType represents different types of notifications 

const (
	NotificationTypeTradeExecution   NotificationType = "trade_execution"
	NotificationTypePnLAlert         NotificationType = "pnl_alert"
	NotificationTypeCircuitBreaker   NotificationType = "circuit_breaker"
	NotificationTypeConsensusFailure NotificationType = "consensus_failure"
	NotificationTypePositionClosed   NotificationType = "position_closed"
	NotificationTypeSafetyGuard      NotificationType = "safety_guard"
	NotificationTypeSystemError      NotificationType = "system_error"
	NotificationTypeDailySummary     NotificationType = "daily_summary"
)

type Platform 

type Platform string

Platform represents the device platform 

const (
	PlatformIOS     Platform = "ios"
	PlatformAndroid Platform = "android"
	PlatformWeb     Platform = "web"
)

type Preferences 

type Preferences struct {
	TradeExecutions   bool `json:"trade_executions"`
	PnLAlerts         bool `json:"pnl_alerts"`
	CircuitBreaker    bool `json:"circuit_breaker"`
	ConsensusFailures bool `json:"consensus_failures"`
	PositionClosed    bool `json:"position_closed"`
	SafetyGuard       bool `json:"safety_guard"`
	SystemErrors      bool `json:"system_errors"`
	DailySummary      bool `json:"daily_summary"`
}

Preferences represents user notification preferences

func DefaultPreferences 

func DefaultPreferences() Preferences

DefaultPreferences returns the default notification preferences

func (Preferences) IsEnabled 

func (p Preferences) IsEnabled(notifType NotificationType) bool

IsEnabled checks if a specific notification type is enabled

type Priority 

type Priority int

Priority represents notification priority levels 

const (
	PriorityInfo      Priority = iota // Informational
	PriorityWarning                   // Warning
	PriorityCritical                  // Critical
	PriorityEmergency                 // Emergency - always delivered
)

func (Priority) String 

func (p Priority) String() string

type Service 

type Service interface {
	// SendToUser sends a notification to all enabled devices for a user
	SendToUser(ctx context.Context, userID string, notification Notification) error

	// SendToDevice sends a notification to a specific device
	SendToDevice(ctx context.Context, deviceToken string, notification Notification) error

	// RegisterDevice registers a new device for push notifications
	RegisterDevice(ctx context.Context, userID, deviceToken string, platform Platform) error

	// UnregisterDevice removes a device token
	UnregisterDevice(ctx context.Context, deviceToken string) error

	// GetUserDevices returns all enabled devices for a user
	GetUserDevices(ctx context.Context, userID string) ([]Device, error)

	// UpdatePreferences updates user notification preferences
	UpdatePreferences(ctx context.Context, userID string, prefs Preferences) error

	// GetPreferences returns user notification preferences
	GetPreferences(ctx context.Context, userID string) (Preferences, error)

	// UpdateDeviceLastUsed updates the last used timestamp for a device
	UpdateDeviceLastUsed(ctx context.Context, deviceToken string) error
}

Service defines the interface for notification operations

type SlackAttachment 

type SlackAttachment struct {
	Color      string       `json:"color,omitempty"`
	Pretext    string       `json:"pretext,omitempty"`
	Title      string       `json:"title,omitempty"`
	TitleLink  string       `json:"title_link,omitempty"`
	Text       string       `json:"text,omitempty"`
	Fields     []SlackField `json:"fields,omitempty"`
	Footer     string       `json:"footer,omitempty"`
	FooterIcon string       `json:"footer_icon,omitempty"`
	Timestamp  int64        `json:"ts,omitempty"`
	MarkdownIn []string     `json:"mrkdwn_in,omitempty"`
}

SlackAttachment represents a Slack message attachment

type SlackBackend 

type SlackBackend struct {
	// contains filtered or unexported fields
}

SlackBackend implements the Backend interface for sending Slack notifications

func NewSlackBackend 

func NewSlackBackend(config SlackConfig) (*SlackBackend, error)

NewSlackBackend creates a new Slack notification backend If webhook URL is empty, creates a mock backend

func (*SlackBackend) Close 

func (s *SlackBackend) Close() error

Close closes the Slack backend

func (*SlackBackend) GetChannel 

func (s *SlackBackend) GetChannel() string

GetChannel returns the default channel

func (*SlackBackend) IsMock 

func (s *SlackBackend) IsMock() bool

IsMock returns true if this is a mock backend

func (*SlackBackend) Name 

func (s *SlackBackend) Name() string

Name returns the backend name

func (*SlackBackend) Send 

func (s *SlackBackend) Send(ctx context.Context, deviceToken string, notification Notification) error

Send sends a notification via Slack webhook

func (*SlackBackend) SetChannel 

func (s *SlackBackend) SetChannel(channel string)

SetChannel sets the default channel

type SlackBlock 

type SlackBlock struct {
	Type     string           `json:"type"`
	Text     *SlackBlockText  `json:"text,omitempty"`
	Elements []SlackElement   `json:"elements,omitempty"`
	Fields   []SlackBlockText `json:"fields,omitempty"`
}

SlackBlock represents a Slack Block Kit block

type SlackBlockText 

type SlackBlockText struct {
	Type  string `json:"type"` // "plain_text" or "mrkdwn"
	Text  string `json:"text"`
	Emoji bool   `json:"emoji,omitempty"`
}

SlackBlockText represents text in a Slack block

type SlackChannelConfig 

type SlackChannelConfig struct {
	WebhookURL string
}

SlackChannelConfig holds config for a Slack notification channel

type SlackConfig 

type SlackConfig struct {
	// WebhookURL is the Slack incoming webhook URL
	WebhookURL string `mapstructure:"webhook_url"`

	// Channel overrides the default channel (optional, use webhook default if empty)
	Channel string `mapstructure:"channel"`

	// Username for the bot (optional)
	Username string `mapstructure:"username"`

	// IconEmoji for the bot (optional, e.g., ":robot_face:")
	IconEmoji string `mapstructure:"icon_emoji"`

	// IconURL for the bot (optional, overrides IconEmoji)
	IconURL string `mapstructure:"icon_url"`

	// Rate limiting
	RateLimitPerMinute int `mapstructure:"rate_limit_per_minute"` // Max messages per minute (0 = unlimited)
	RetryAttempts      int `mapstructure:"retry_attempts"`        // Number of retry attempts on failure
	RetryDelaySeconds  int `mapstructure:"retry_delay_seconds"`   // Delay between retries

	// Timeout for HTTP requests
	TimeoutSeconds int `mapstructure:"timeout_seconds"`
}

SlackConfig holds Slack webhook configuration

func DefaultSlackConfig 

func DefaultSlackConfig() SlackConfig

DefaultSlackConfig returns default Slack configuration

type SlackElement 

type SlackElement struct {
	Type string `json:"type"`
	Text string `json:"text,omitempty"`
}

SlackElement represents an element in a Slack block

type SlackField 

type SlackField struct {
	Title string `json:"title"`
	Value string `json:"value"`
	Short bool   `json:"short"`
}

SlackField represents a field in a Slack attachment

type SlackMessage 

type SlackMessage struct {
	Channel     string            `json:"channel,omitempty"`
	Username    string            `json:"username,omitempty"`
	IconEmoji   string            `json:"icon_emoji,omitempty"`
	IconURL     string            `json:"icon_url,omitempty"`
	Text        string            `json:"text,omitempty"`
	Attachments []SlackAttachment `json:"attachments,omitempty"`
	Blocks      []SlackBlock      `json:"blocks,omitempty"`
}

SlackMessage represents a Slack webhook message

type TelegramChannel 

type TelegramChannel struct {
	// contains filtered or unexported fields
}

TelegramChannel sends notifications via Telegram Bot API

func NewTelegramChannel 

func NewTelegramChannel(cfg TelegramChannelConfig) *TelegramChannel

NewTelegramChannel creates a new Telegram notification channel

func (*TelegramChannel) Close 

func (t *TelegramChannel) Close() error

func (*TelegramChannel) Name 

func (t *TelegramChannel) Name() string

func (*TelegramChannel) Send 

func (t *TelegramChannel) Send(ctx context.Context, event Event) error

Send formats and sends a notification event via Telegram

type TelegramChannelConfig 

type TelegramChannelConfig struct {
	BotToken   string
	ChatID     string
	HTTPClient *http.Client
	BaseURL    string // override for testing
}

TelegramChannelConfig holds Telegram channel configuration

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.