price pipeline

Signed-off-by: GnomeZworc <nicolas.boufidjeline@g3e.fr>

internal/store/instrument.go

@@ -0,0 +1,78 @@
+package store
+
+import (
+	"context"
+
+	"github.com/jackc/pgx/v5"
+)
+
+type Instrument struct {
+	ID             int32  `json:"id"`
+	Type           string `json:"type"`
+	Code           string `json:"code"`
+	Name           string `json:"name"`
+	DeviseCotation string `json:"devise_cotation"`
+}
+
+type CreateInstrumentParams struct {
+	Type           string `json:"type"`
+	Code           string `json:"code"`
+	Name           string `json:"name"`
+	DeviseCotation string `json:"devise_cotation"`
+}
+
+func (s *Store) ListInstruments(ctx context.Context) ([]Instrument, error) {
+	rows, err := s.pool.Query(ctx,
+		`SELECT id, type, code, name, devise_cotation FROM instrument ORDER BY type, code`)
+	if err != nil {
+		return nil, err
+	}
+	return pgx.CollectRows(rows, pgx.RowToStructByName[Instrument])
+}
+
+func (s *Store) GetInstrument(ctx context.Context, id int32) (Instrument, error) {
+	rows, err := s.pool.Query(ctx,
+		`SELECT id, type, code, name, devise_cotation FROM instrument WHERE id = $1`, id)
+	if err != nil {
+		return Instrument{}, err
+	}
+	return pgx.CollectOneRow(rows, pgx.RowToStructByName[Instrument])
+}
+
+func (s *Store) GetInstrumentByCode(ctx context.Context, code string) (Instrument, error) {
+	rows, err := s.pool.Query(ctx,
+		`SELECT id, type, code, name, devise_cotation FROM instrument WHERE code = $1`, code)
+	if err != nil {
+		return Instrument{}, err
+	}
+	return pgx.CollectOneRow(rows, pgx.RowToStructByName[Instrument])
+}
+
+func (s *Store) CreateInstrument(ctx context.Context, p CreateInstrumentParams) (Instrument, error) {
+	rows, err := s.pool.Query(ctx,
+		`INSERT INTO instrument (type, code, name, devise_cotation)
+		 VALUES ($1, $2, $3, $4)
+		 RETURNING id, type, code, name, devise_cotation`,
+		p.Type, p.Code, p.Name, p.DeviseCotation)
+	if err != nil {
+		return Instrument{}, err
+	}
+	return pgx.CollectOneRow(rows, pgx.RowToStructByName[Instrument])
+}
+
+func (s *Store) UpdateInstrument(ctx context.Context, id int32, p CreateInstrumentParams) (Instrument, error) {
+	rows, err := s.pool.Query(ctx,
+		`UPDATE instrument SET type = $2, code = $3, name = $4, devise_cotation = $5
+		 WHERE id = $1
+		 RETURNING id, type, code, name, devise_cotation`,
+		id, p.Type, p.Code, p.Name, p.DeviseCotation)
+	if err != nil {
+		return Instrument{}, err
+	}
+	return pgx.CollectOneRow(rows, pgx.RowToStructByName[Instrument])
+}
+
+func (s *Store) DeleteInstrument(ctx context.Context, id int32) error {
+	_, err := s.pool.Exec(ctx, `DELETE FROM instrument WHERE id = $1`, id)
+	return err
+}

internal/store/price_history.go

@@ -0,0 +1,41 @@
+package store
+
+import (
+	"context"
+	"time"
+)
+
+func (s *Store) UpsertPrice(ctx context.Context, instrumentID int32, fetchedAt time.Time, prix float64) error {
+	_, err := s.pool.Exec(ctx,
+		`INSERT INTO price_history (instrument_id, fetched_at, prix)
+		 VALUES ($1, $2, $3)
+		 ON CONFLICT (instrument_id, fetched_at) DO UPDATE SET prix = EXCLUDED.prix`,
+		instrumentID, fetchedAt, prix)
+	return err
+}
+
+// GetLastPrice retourne le dernier prix connu pour un instrument.
+func (s *Store) GetLastPrice(ctx context.Context, instrumentID int32) (float64, time.Time, error) {
+	var prix float64
+	var fetchedAt time.Time
+	err := s.pool.QueryRow(ctx,
+		`SELECT prix, fetched_at FROM price_history
+		 WHERE instrument_id = $1
+		 ORDER BY fetched_at DESC LIMIT 1`,
+		instrumentID).Scan(&prix, &fetchedAt)
+	return prix, fetchedAt, err
+}
+
+// CleanPastDays supprime pour chaque jour passé toutes les entrées sauf la dernière par instrument.
+func (s *Store) CleanPastDays(ctx context.Context) error {
+	_, err := s.pool.Exec(ctx, `
+		DELETE FROM price_history ph
+		WHERE ph.fetched_at::date < CURRENT_DATE
+		  AND ph.fetched_at < (
+		      SELECT MAX(ph2.fetched_at)
+		      FROM price_history ph2
+		      WHERE ph2.instrument_id = ph.instrument_id
+		        AND ph2.fetched_at::date = ph.fetched_at::date
+		  )`)
+	return err
+}

internal/store/store.go

@@ -0,0 +1,11 @@
+package store
+
+import "github.com/jackc/pgx/v5/pgxpool"
+
+type Store struct {
+	pool *pgxpool.Pool
+}
+
+func New(pool *pgxpool.Pool) *Store {
+	return &Store{pool: pool}
+}

internal/store/ticker_cache.go

@@ -0,0 +1,31 @@
+package store
+
+import (
+	"context"
+
+	"github.com/jackc/pgx/v5"
+)
+
+// GetTicker retourne le ticker (ou CoinGecko ID) mis en cache pour un instrument.
+// Retourne pgx.ErrNoRows si absent.
+func (s *Store) GetTicker(ctx context.Context, instrumentID int32) (string, error) {
+	var ticker string
+	err := s.pool.QueryRow(ctx,
+		`SELECT ticker FROM instrument_ticker_cache WHERE instrument_id = $1`,
+		instrumentID).Scan(&ticker)
+	return ticker, err
+}
+
+// UpsertTicker met à jour ou insère le ticker pour un instrument.
+func (s *Store) UpsertTicker(ctx context.Context, instrumentID int32, ticker string) error {
+	_, err := s.pool.Exec(ctx,
+		`INSERT INTO instrument_ticker_cache (instrument_id, ticker, fetched_at)
+		 VALUES ($1, $2, NOW())
+		 ON CONFLICT (instrument_id) DO UPDATE SET ticker = EXCLUDED.ticker, fetched_at = NOW()`,
+		instrumentID, ticker)
+	return err
+}
+
+func isNotFound(err error) bool {
+	return err == pgx.ErrNoRows
+}