yash/crypto_alert_bot
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

alert service

Browse source
This commit is contained in:
yash 2026-02-25 23:52:21 +03:00
parent 0e73841b3e
commit 608561ab38
8 changed files with 283 additions and 8 deletions
Download patch file Download diff file Expand all files Collapse all files

8
internal/entities/alert.go
View file

@ -4,9 +4,17 @@ import "github.com/shopspring/decimal"
type AlertID string
type AlertCondition string
const (
AlertConditionAbove AlertCondition = "above" // trigger when price rises to target
AlertConditionBelow AlertCondition = "below" // trigger when price drops to target
)
type Alert struct {
ID AlertID
UserID UserID
Price decimal.Decimal
Condition AlertCondition
Instrument Instrument
}

64
internal/repository/postgresql/alert.go
View file

@ -9,14 +9,14 @@ import (
)
const saveAlertQuery = `
insert into alert(user_id, instrument_id, price)
values ($1, $2, $3)
insert into alert(user_id, instrument_id, price, condition)
values ($1, $2, $3, $4)
returning id`
func (p *Postgresql) SaveAlert(ctx context.Context, alert *entities.Alert) (entities.AlertID, error) {
var id entities.AlertID
err := p.db.QueryRow(ctx, saveAlertQuery, alert.UserID, alert.Instrument.ID, alert.Price).Scan(&id)
err := p.db.QueryRow(ctx, saveAlertQuery, alert.UserID, alert.Instrument.ID, alert.Price.String(), alert.Condition).Scan(&id)
if err != nil {
return "", fmt.Errorf("failed to exec saveAlertQuery: %w", err)
}
@ -24,8 +24,47 @@ func (p *Postgresql) SaveAlert(ctx context.Context, alert *entities.Alert) (enti
return id, nil
}
const allActiveAlertsQuery = `
select a.id, a.user_id, a.price, a.condition, i.id, c_base.symbol, c_quote.symbol
from alert a
join instrument i on i.id = a.instrument_id
join currency c_base on c_base.id = i.base_currency_id
join currency c_quote on c_quote.id = i.quoted_currency_id
where a.active = true
order by a.id`
func (p *Postgresql) AllActiveAlerts(ctx context.Context) ([]entities.Alert, error) {
rows, err := p.db.Query(ctx, allActiveAlertsQuery)
if err != nil {
return nil, fmt.Errorf("failed to exec allActiveAlertsQuery: %w", err)
}
defer rows.Close()
var alerts []entities.Alert
for rows.Next() {
var alert entities.Alert
var priceStr string
if err := rows.Scan(
&alert.ID, &alert.UserID, &priceStr, &alert.Condition,
&alert.Instrument.ID, &alert.Instrument.BaseCurrency, &alert.Instrument.QuoteCurrency,
); err != nil {
return nil, fmt.Errorf("failed to scan alert row: %w", err)
}
alert.Price, err = decimal.NewFromString(priceStr)
if err != nil {
return nil, fmt.Errorf("failed to parse alert price: %w", err)
}
alerts = append(alerts, alert)
}
return alerts, nil
}
const alertByIDQuery = `
select a.id, a.user_id, a.price, i.id, c_base.symbol, c_quote.symbol
select a.id, a.user_id, a.price, a.condition, i.id, c_base.symbol, c_quote.symbol
from alert a
join instrument i on i.id = a.instrument_id
join currency c_base on c_base.id = i.base_currency_id
@ -37,7 +76,7 @@ func (p *Postgresql) AlertByID(ctx context.Context, id entities.AlertID) (*entit
var priceStr string
err := p.db.QueryRow(ctx, alertByIDQuery, id).Scan(
&alert.ID, &alert.UserID, &priceStr,
&alert.ID, &alert.UserID, &priceStr, &alert.Condition,
&alert.Instrument.ID, &alert.Instrument.BaseCurrency, &alert.Instrument.QuoteCurrency,
)
if err != nil {
@ -53,7 +92,7 @@ func (p *Postgresql) AlertByID(ctx context.Context, id entities.AlertID) (*entit
}
const alertsByUserIDQuery = `
select a.id, a.user_id, a.price, i.id, c_base.symbol, c_quote.symbol
select a.id, a.user_id, a.price, a.condition, i.id, c_base.symbol, c_quote.symbol
from alert a
join instrument i on i.id = a.instrument_id
join currency c_base on c_base.id = i.base_currency_id
@ -75,7 +114,7 @@ func (p *Postgresql) AlertsByUserID(ctx context.Context, userID entities.UserID,
var priceStr string
if err := rows.Scan(
&alert.ID, &alert.UserID, &priceStr,
&alert.ID, &alert.UserID, &priceStr, &alert.Condition,
&alert.Instrument.ID, &alert.Instrument.BaseCurrency, &alert.Instrument.QuoteCurrency,
); err != nil {
return nil, fmt.Errorf("failed to scan alert row: %w", err)
@ -103,6 +142,17 @@ func (p *Postgresql) DeleteAlert(ctx context.Context, id entities.AlertID) error
return nil
}
const disableAlertQuery = "update alert set active = false where id = $1"
func (p *Postgresql) DisableAlert(ctx context.Context, id entities.AlertID) error {
_, err := p.db.Exec(ctx, disableAlertQuery, id)
if err != nil {
return fmt.Errorf("failed to exec disableAlertQuery: %w", err)
}
return nil
}
const updateAlertPriceQuery = "update alert set price = $2 where id = $1"
func (p *Postgresql) UpdateAlertPrice(ctx context.Context, id entities.AlertID, price decimal.Decimal) error {

1
internal/repository/postgresql/migrations/000001_init.down.sql
View file

@ -1,4 +1,5 @@
drop table if exists alert;
drop type alert_condition;
drop table if exists instrument;
drop table if exists currency;
drop table if exists users;

5
internal/repository/postgresql/migrations/000001_init.up.sql
View file

@ -17,12 +17,15 @@ create table if not exists instrument (
UNIQUE (base_currency_id, quoted_currency_id)
);
create type alert_condition as enum ('above', 'below');
create table if not exists alert (
id uuid primary key not null default gen_random_uuid(),
user_id uuid references users(id) not null,
instrument_id uuid references instrument(id) not null,
price text not null,
active bool not null default true
active bool not null default true,
condition alert_condition not null
);
insert into currency(symbol) values ('USDT'), ('BTC'), ('ETH'), ('SOL');

2
internal/repository/repository.go
View file

@ -16,8 +16,10 @@ type Storage interface {
CreateInstrument(ctx context.Context, instrument *entities.Instrument) (entities.InstrumentID, error)
SaveAlert(ctx context.Context, alert *entities.Alert) (entities.AlertID, error)
AllActiveAlerts(ctx context.Context) ([]entities.Alert, error)
AlertByID(ctx context.Context, id entities.AlertID) (*entities.Alert, error)
AlertsByUserID(ctx context.Context, userID entities.UserID, offset, limit int) ([]entities.Alert, error)
DeleteAlert(ctx context.Context, id entities.AlertID) error
DisableAlert(ctx context.Context, id entities.AlertID) error
UpdateAlertPrice(ctx context.Context, id entities.AlertID, price decimal.Decimal) error
}

131
internal/service/alerter/alerter.go Normal file
View file

@ -0,0 +1,131 @@
package alerter
import (
"context"
"fmt"
"log/slog"
"time"
"gitea.computernetthings.ru/yash/crypto_alert_bot/internal/entities"
"gitea.computernetthings.ru/yash/crypto_alert_bot/internal/provider"
"github.com/shopspring/decimal"
)
type Notifier interface {
NotifyAlert(ctx context.Context, userID entities.UserID, alert *entities.Alert, currentPrice decimal.Decimal) error
}
type Storage interface {
AllActiveAlerts(ctx context.Context) ([]entities.Alert, error)
DisableAlert(ctx context.Context, id entities.AlertID) error
}
type Alerter struct {
log *slog.Logger
cache *alertsCache
priceProvider provider.Provider
notifier Notifier
storage Storage
}
const interval = time.Minute
func New(log *slog.Logger, priceProvider provider.Provider, notifier Notifier, storage Storage) *Alerter {
return &Alerter{
log: log,
cache: newCache(),
priceProvider: priceProvider,
notifier: notifier,
storage: storage,
}
}
func (a *Alerter) LoadAlerts(ctx context.Context) error {
alerts, err := a.storage.AllActiveAlerts(ctx)
if err != nil {
return fmt.Errorf("failed to load alerts: %w", err)
}
for i := range alerts {
a.cache.Add(&alerts[i])
}
a.log.Info("alerts loaded", "count", len(alerts))
return nil
}
func (a *Alerter) AddAlert(alert *entities.Alert) {
a.cache.Add(alert)
}
func (a *Alerter) RemoveAlert(id entities.AlertID) {
a.cache.Remove(id)
}
func (a *Alerter) Run(ctx context.Context) {
go func() {
ticker := time.NewTicker(interval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
}
a.log.Info("start checking alerts")
if err := a.checkAlerts(ctx); err != nil {
a.log.Error("failed to check alerts", "err", err)
continue
}
a.log.Info("alerts checked")
}
}()
}
// TODO: parallel checking for different instruments.
func (a *Alerter) checkAlerts(ctx context.Context) error {
instruments := a.cache.Instruments()
for _, instrument := range instruments {
price, err := a.priceProvider.Price(ctx, instrument)
if err != nil {
a.log.Error("failed to get price", "instrument", instrument.ID, "err", err)
continue
}
alerts := a.cache.AlertsByInstrument(instrument.ID)
for _, alert := range alerts {
switch alert.Condition {
case entities.AlertConditionAbove:
if price.Ask.GreaterThanOrEqual(alert.Price) {
a.triggerAlert(ctx, alert, price.Ask)
}
case entities.AlertConditionBelow:
if price.Bid.LessThanOrEqual(alert.Price) {
a.triggerAlert(ctx, alert, price.Bid)
}
}
}
}
return nil
}
func (a *Alerter) triggerAlert(ctx context.Context, alert *entities.Alert, currentPrice decimal.Decimal) {
if err := a.notifier.NotifyAlert(ctx, alert.UserID, alert, currentPrice); err != nil {
a.log.Error("failed to notify alert", "alert_id", alert.ID, "err", err)
return
}
a.cache.Remove(alert.ID)
if err := a.storage.DisableAlert(ctx, alert.ID); err != nil {
a.log.Error("failed to disable alert in db", "alert_id", alert.ID, "err", err)
}
}

71
internal/service/alerter/cache.go Normal file
View file

@ -0,0 +1,71 @@
package alerter
import (
"sync"
"gitea.computernetthings.ru/yash/crypto_alert_bot/internal/entities"
)
type alertsCache struct {
mu sync.RWMutex
byID map[entities.AlertID]*entities.Alert
byInstrument map[entities.InstrumentID]map[entities.AlertID]*entities.Alert
}
func newCache() *alertsCache {
return &alertsCache{
byID: make(map[entities.AlertID]*entities.Alert),
byInstrument: make(map[entities.InstrumentID]map[entities.AlertID]*entities.Alert),
}
}
func (c *alertsCache) Add(a *entities.Alert) {
c.mu.Lock()
defer c.mu.Unlock()
c.byID[a.ID] = a
if _, ok := c.byInstrument[a.Instrument.ID]; !ok {
c.byInstrument[a.Instrument.ID] = make(map[entities.AlertID]*entities.Alert)
}
c.byInstrument[a.Instrument.ID][a.ID] = a
}
func (c *alertsCache) Remove(id entities.AlertID) {
c.mu.Lock()
defer c.mu.Unlock()
a, ok := c.byID[id]
if !ok {
return
}
delete(c.byID, id)
delete(c.byInstrument[a.Instrument.ID], id)
}
func (c *alertsCache) AlertsByInstrument(id entities.InstrumentID) []*entities.Alert {
c.mu.RLock()
defer c.mu.RUnlock()
alerts := c.byInstrument[id]
result := make([]*entities.Alert, 0, len(alerts))
for _, a := range alerts {
result = append(result, a)
}
return result
}
func (c *alertsCache) Instruments() []entities.Instrument {
c.mu.RLock()
defer c.mu.RUnlock()
instruments := make([]entities.Instrument, 0, len(c.byInstrument))
for _, alerts := range c.byInstrument {
for _, a := range alerts {
instruments = append(instruments, a.Instrument)
break
}
}
return instruments
}

9
internal/usecase/alert.go
View file

@ -55,3 +55,12 @@ func (uc *Usecase) UpdateAlertPrice(ctx context.Context, alertID entities.AlertI
return nil
}
func (uc *Usecase) DisableAlert(ctx context.Context, alertID entities.AlertID) error {
if err := uc.storage.DisableAlert(ctx, alertID); err != nil {
uc.log.Error("failed to disable alert", "alert_id", alertID, "err", err)
return fmt.Errorf("failed to disable alert: %w", err)
}
return nil
}