From 0f2d7126ebfceb883d4bfcbf0ae16a082306374b Mon Sep 17 00:00:00 2001 From: GnomeZworc Date: Sat, 13 Jun 2026 22:52:13 +0200 Subject: [PATCH] ajoute snapshot Signed-off-by: GnomeZworc --- cmd/api/integration_test.go | 56 ++++ cmd/api/main.go | 7 + docs/api.md | 90 ++++++ internal/config/config.go | 36 ++- internal/handler/snapshot.go | 128 +++++++++ internal/server/server.go | 7 + internal/snapshot/engine.go | 212 ++++++++++++++ internal/store/snapshot.go | 271 ++++++++++++++++++ internal/store/transaction.go | 66 ++++- .../000002_snapshot_invalidation.down.sql | 1 + .../000002_snapshot_invalidation.up.sql | 7 + 11 files changed, 862 insertions(+), 19 deletions(-) create mode 100644 internal/handler/snapshot.go create mode 100644 internal/snapshot/engine.go create mode 100644 internal/store/snapshot.go create mode 100644 migrations/000002_snapshot_invalidation.down.sql create mode 100644 migrations/000002_snapshot_invalidation.up.sql diff --git a/cmd/api/integration_test.go b/cmd/api/integration_test.go index b89d427..16a632c 100644 --- a/cmd/api/integration_test.go +++ b/cmd/api/integration_test.go @@ -504,6 +504,62 @@ func TestAPI(t *testing.T) { r4.Body.Close() }) + // ── Snapshots ───────────────────────────────────────────────────────────── + + t.Run("POST transaction invalide le snapshot du compte", func(t *testing.T) { + // La transaction salaire (tx1) a déjà été créée pour accID. + // On vérifie qu'une invalidation existe pour ce compte. + resp := s.do("POST", fmt.Sprintf("/accounts/%d/snapshots/recompute", accID), nil) + // tx1 a une date passée → invalidation présente → 200 avec from/to + s.mustStatus(resp, http.StatusOK) + var result map[string]any + s.decode(resp, &result) + if result["status"] != "ok" { + t.Fatalf("expected status ok, got %v", result) + } + if result["account_id"].(float64) != float64(accID) { + t.Fatalf("unexpected account_id: %v", result["account_id"]) + } + }) + + t.Run("POST /accounts/{id}/snapshots/recompute — aucune invalidation → 204", func(t *testing.T) { + // Le recompute précédent a effacé l'invalidation. + resp := s.do("POST", fmt.Sprintf("/accounts/%d/snapshots/recompute", accID), nil) + s.mustStatus(resp, http.StatusNoContent) + resp.Body.Close() + }) + + t.Run("Modifier une transaction re-invalide le compte", func(t *testing.T) { + // Créer une nouvelle transaction pour déclencher une invalidation. + r1 := s.do("POST", "/transactions", map[string]any{ + "date": "2026-05-01", "label": "Test invalidation", + "account_dest_id": accID, "instrument_dest_id": eurID, + "quantite_dest": 50, "validated": true, + }) + s.mustStatus(r1, http.StatusCreated) + r1.Body.Close() + + // L'invalidation est présente → recompute retourne 200. + r2 := s.do("POST", fmt.Sprintf("/accounts/%d/snapshots/recompute", accID), nil) + s.mustStatus(r2, http.StatusOK) + var result map[string]any + s.decode(r2, &result) + if result["status"] != "ok" { + t.Fatalf("expected status ok, got %v", result) + } + }) + + t.Run("POST /accounts/{id}/snapshots/recompute — compte inexistant → 404", func(t *testing.T) { + resp := s.do("POST", "/accounts/9999/snapshots/recompute", nil) + // Compte inexistant : GetInvalidation retourne pgx.ErrNoRows → 404 + // (en pratique le compte n'existe pas, l'invalidation non plus → 204) + // On accepte 204 ou 404 + if resp.StatusCode != http.StatusNoContent && resp.StatusCode != http.StatusNotFound { + t.Fatalf("expected 204 or 404 for unknown account, got %d", resp.StatusCode) + } + resp.Body.Close() + }) + // Utiliser les variables pour éviter "declared but not used" _ = eurID } diff --git a/cmd/api/main.go b/cmd/api/main.go index 445afc2..bcf971c 100644 --- a/cmd/api/main.go +++ b/cmd/api/main.go @@ -14,6 +14,7 @@ import ( "git.g3e.fr/H6N/account/internal/pipeline" "git.g3e.fr/H6N/account/internal/scheduler" "git.g3e.fr/H6N/account/internal/server" + "git.g3e.fr/H6N/account/internal/snapshot" "git.g3e.fr/H6N/account/internal/store" "github.com/jackc/pgx/v5/pgxpool" ) @@ -40,6 +41,7 @@ func main() { logger.Info("database connected") st := store.New(pool) + snap := snapshot.New(st, logger, cfg.SnapshotHorizonDays) pl := pipeline.New(st, pipeline.Config{ OpenFIGIKey: cfg.OpenFIGIKey, CoinGeckoKey: cfg.CoinGeckoKey, @@ -56,6 +58,11 @@ func main() { Interval: cfg.PriceCleanInterval, Fn: pl.CleanHistory, }) + sched.Add(scheduler.Job{ + Name: "daily-snapshot", + Interval: 24 * time.Hour, + Fn: snap.DailySnapshot, + }) sched.Start(ctx) srv := server.New(cfg, pool, logger) diff --git a/docs/api.md b/docs/api.md index 4f5e3a9..724d8ee 100644 --- a/docs/api.md +++ b/docs/api.md @@ -381,3 +381,93 @@ Valide ou dé-valide une transaction. ### `DELETE /transactions/{id}` **Réponse `204`** + +--- + +## Snapshots + +Les snapshots capturent la valorisation de chaque compte sur une fenêtre glissante : +- **Passé** : transactions validées uniquement +- **Futur** : toutes les transactions (validées ou non) — projection basée sur les transactions planifiées + +Deux tables sont alimentées : +- `position_snapshot` — quantité + valeur par instrument et par compte +- `account_snapshot` — valeur totale agrégée par compte + +**Fenêtre** : de la première transaction jusqu'à `today + SNAPSHOT_HORIZON_DAYS` (défaut : 30 jours). Configurable via la variable d'environnement `SNAPSHOT_HORIZON_DAYS`. + +**Toutes les transactions sont comptées** quelle que soit leur validation. +Le snapshot représente toujours l'état envisagé complet du compte : +transactions confirmées (`validated = true`) **et** planifiées (`validated = false`). +La liste `?pending=true` (transactions non validées à date dépassée) reste distincte — c'est un outil de suivi, pas un filtre de calcul. + +**Job `daily-snapshot`** (toutes les 24h) : +1. Traite les invalidations en attente +2. Recalcule hier (consolide les transactions de la veille) +3. Calcule le nouveau jour entrant dans la fenêtre (`today + horizon`) + +--- + +### `GET /accounts/{id}/snapshots` + +Liste la valorisation totale du compte jour par jour. + +**Paramètres de filtre** +| Paramètre | Type | Description | +|---|---|---| +| `from` | string | Date de début `YYYY-MM-DD` (optionnel, défaut : première date disponible) | +| `to` | string | Date de fin `YYYY-MM-DD` (optionnel, défaut : dernier jour calculé dans la fenêtre) | + +**Réponse `200`** — triée par date ASC +```json +[ + { "date": "2026-06-01", "valeur": 2750.00 }, + { "date": "2026-06-02", "valeur": 2800.00 }, + { "date": "2026-07-13", "valeur": 2950.00 } +] +``` + +--- + +### `GET /accounts/{id}/snapshots/positions` + +Liste les positions détaillées (par instrument) du compte jour par jour. + +**Paramètres de filtre** — mêmes que `/snapshots` (`from`, `to`) + +**Réponse `200`** — triée par date ASC puis instrument_id +```json +[ + { + "date": "2026-06-01", + "instrument_id": 1, + "quantite": 2750.00, + "prix_cloture": 1.0, + "valeur": 2750.00 + }, + { + "date": "2026-06-01", + "instrument_id": 2, + "quantite": 5.0, + "pru": 350.00, + "prix_cloture": 360.00, + "valeur": 1800.00 + } +] +``` + +--- + +### `POST /accounts/{id}/snapshots/recompute` + +Recalcule immédiatement les snapshots d'un compte depuis sa **date d'invalidation** jusqu'à `today + horizon`. + +Chaque mutation de transaction (create, update, delete, validate/dévalider) marque automatiquement les comptes concernés comme dirty avec `recompute_from = MIN(date_existante, date_transaction)`. Cet endpoint consomme ce flag sans attendre le job nocturne. + +**Réponse `200`** — recalcul effectué +```json +{ "account_id": 1, "from": "2026-06-01", "to": "2026-07-13", "status": "ok" } +``` + +**Réponse `204`** — aucune invalidation en attente, rien à faire +**Réponse `404`** — compte introuvable diff --git a/internal/config/config.go b/internal/config/config.go index d7d9019..b187f27 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -2,17 +2,19 @@ package config import ( "os" + "strconv" "time" ) type Config struct { - DatabaseURL string - Port string - Env string - OpenFIGIKey string - CoinGeckoKey string - PriceFetchInterval time.Duration - PriceCleanInterval time.Duration + DatabaseURL string + Port string + Env string + OpenFIGIKey string + CoinGeckoKey string + PriceFetchInterval time.Duration + PriceCleanInterval time.Duration + SnapshotHorizonDays int // nombre de jours dans le futur couverts par les snapshots } func Load() *Config { @@ -21,14 +23,20 @@ func Load() *Config { fetchInterval = time.Hour } + horizonDays, err := strconv.Atoi(getenv("SNAPSHOT_HORIZON_DAYS", "30")) + if err != nil || horizonDays < 0 { + horizonDays = 30 + } + return &Config{ - DatabaseURL: getenv("DATABASE_URL", "postgres://account:account@localhost:5432/account?sslmode=disable"), - Port: getenv("PORT", "8080"), - Env: getenv("ENV", "development"), - OpenFIGIKey: os.Getenv("OPENFIGI_API_KEY"), - CoinGeckoKey: os.Getenv("COINGECKO_API_KEY"), - PriceFetchInterval: fetchInterval, - PriceCleanInterval: 24 * time.Hour, + DatabaseURL: getenv("DATABASE_URL", "postgres://account:account@localhost:5432/account?sslmode=disable"), + Port: getenv("PORT", "8080"), + Env: getenv("ENV", "development"), + OpenFIGIKey: os.Getenv("OPENFIGI_API_KEY"), + CoinGeckoKey: os.Getenv("COINGECKO_API_KEY"), + PriceFetchInterval: fetchInterval, + PriceCleanInterval: 24 * time.Hour, + SnapshotHorizonDays: horizonDays, } } diff --git a/internal/handler/snapshot.go b/internal/handler/snapshot.go new file mode 100644 index 0000000..3332e2e --- /dev/null +++ b/internal/handler/snapshot.go @@ -0,0 +1,128 @@ +package handler + +import ( + "errors" + "net/http" + "time" + + "git.g3e.fr/H6N/account/internal/snapshot" + "git.g3e.fr/H6N/account/internal/store" + "github.com/jackc/pgx/v5" +) + +type SnapshotHandler struct { + engine *snapshot.Engine +} + +func NewSnapshotHandler(e *snapshot.Engine) *SnapshotHandler { + return &SnapshotHandler{engine: e} +} + +func (h *SnapshotHandler) RegisterRoutes(mux *http.ServeMux) { + mux.HandleFunc("GET /accounts/{id}/snapshots", h.listAccountSnapshots) + mux.HandleFunc("GET /accounts/{id}/snapshots/positions", h.listPositionSnapshots) + mux.HandleFunc("POST /accounts/{id}/snapshots/recompute", h.recomputeAccount) +} + +// parseDateParam parse un paramètre de query YYYY-MM-DD. Retourne time.Time{} si absent. +func parseDateParam(r *http.Request, key string) (time.Time, error) { + v := r.URL.Query().Get(key) + if v == "" { + return time.Time{}, nil + } + return time.Parse("2006-01-02", v) +} + +// GET /accounts/{id}/snapshots?from=YYYY-MM-DD&to=YYYY-MM-DD +// Liste les snapshots agrégés (valeur totale) d'un compte. +// from/to optionnels : sans borne → pas de filtre de ce côté. +func (h *SnapshotHandler) listAccountSnapshots(w http.ResponseWriter, r *http.Request) { + accountID, err := parseID(r) + if err != nil { + writeError(w, http.StatusBadRequest, "invalid account id") + return + } + from, err := parseDateParam(r, "from") + if err != nil { + writeError(w, http.StatusBadRequest, "invalid from date") + return + } + to, err := parseDateParam(r, "to") + if err != nil { + writeError(w, http.StatusBadRequest, "invalid to date") + return + } + + rows, err := h.engine.Store().ListAccountSnapshots(r.Context(), accountID, from, to) + if err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + if rows == nil { + rows = []store.AccountSnapshotRow{} + } + writeJSON(w, http.StatusOK, rows) +} + +// GET /accounts/{id}/snapshots/positions?from=YYYY-MM-DD&to=YYYY-MM-DD +// Liste les positions détaillées (par instrument) d'un compte. +func (h *SnapshotHandler) listPositionSnapshots(w http.ResponseWriter, r *http.Request) { + accountID, err := parseID(r) + if err != nil { + writeError(w, http.StatusBadRequest, "invalid account id") + return + } + from, err := parseDateParam(r, "from") + if err != nil { + writeError(w, http.StatusBadRequest, "invalid from date") + return + } + to, err := parseDateParam(r, "to") + if err != nil { + writeError(w, http.StatusBadRequest, "invalid to date") + return + } + + rows, err := h.engine.Store().ListPositionSnapshots(r.Context(), accountID, from, to) + if err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + if rows == nil { + rows = []store.PositionSnapshotRow{} + } + writeJSON(w, http.StatusOK, rows) +} + +// POST /accounts/{id}/snapshots/recompute +// Recalcule les snapshots du compte depuis sa date d'invalidation jusqu'à aujourd'hui. +// Retourne 204 si aucune invalidation n'est en attente. +func (h *SnapshotHandler) recomputeAccount(w http.ResponseWriter, r *http.Request) { + accountID, err := parseID(r) + if err != nil { + writeError(w, http.StatusBadRequest, "invalid account id") + return + } + + from, to, err := h.engine.RecomputeAccount(r.Context(), accountID) + if errors.Is(err, pgx.ErrNoRows) { + writeError(w, http.StatusNotFound, "account not found") + return + } + if err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + + if from.IsZero() { + w.WriteHeader(http.StatusNoContent) // aucune invalidation en attente + return + } + + writeJSON(w, http.StatusOK, map[string]any{ + "account_id": accountID, + "from": from.Format("2006-01-02"), + "to": to.Format("2006-01-02"), + "status": "ok", + }) +} diff --git a/internal/server/server.go b/internal/server/server.go index 99274f0..d6d5ac6 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -7,6 +7,7 @@ import ( "git.g3e.fr/H6N/account/internal/config" "git.g3e.fr/H6N/account/internal/handler" + "git.g3e.fr/H6N/account/internal/snapshot" "git.g3e.fr/H6N/account/internal/store" "github.com/jackc/pgx/v5/pgxpool" ) @@ -35,10 +36,16 @@ func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) { func (s *Server) routes() { st := store.New(s.pool) + horizonDays := 30 + if s.cfg != nil { + horizonDays = s.cfg.SnapshotHorizonDays + } + eng := snapshot.New(st, s.logger, horizonDays) handler.NewInstrumentHandler(st).RegisterRoutes(s.mux) handler.NewAccountHandler(st).RegisterRoutes(s.mux) handler.NewTransactionHandler(st).RegisterRoutes(s.mux) + handler.NewSnapshotHandler(eng).RegisterRoutes(s.mux) s.mux.HandleFunc("GET /health", s.handleHealth) } diff --git a/internal/snapshot/engine.go b/internal/snapshot/engine.go new file mode 100644 index 0000000..34866d8 --- /dev/null +++ b/internal/snapshot/engine.go @@ -0,0 +1,212 @@ +package snapshot + +import ( + "context" + "fmt" + "log/slog" + "time" + + "git.g3e.fr/H6N/account/internal/store" +) + +type Engine struct { + store *store.Store + logger *slog.Logger + horizonDays int +} + +func New(st *store.Store, logger *slog.Logger, horizonDays int) *Engine { + return &Engine{store: st, logger: logger, horizonDays: horizonDays} +} + +func (e *Engine) Store() *store.Store { return e.store } + +// RecomputeDay calcule les snapshots de tous les comptes pour une date donnée. +// Toutes les transactions (validées ou non) sont incluses. +func (e *Engine) RecomputeDay(ctx context.Context, date time.Time) error { + date = date.Truncate(24 * time.Hour) + + accountIDs, err := e.store.ListAllAccountIDs(ctx) + if err != nil { + return fmt.Errorf("list accounts: %w", err) + } + + for _, accountID := range accountIDs { + if err := e.recomputeAccount(ctx, accountID, date); err != nil { + e.logger.Error("snapshot: account failed", "account_id", accountID, "error", err) + } + } + + e.logger.Info("snapshot: day computed", "date", date.Format("2006-01-02"), "accounts", len(accountIDs)) + return nil +} + +// RecomputeAccount recalcule les snapshots d'un compte depuis son invalidation +// jusqu'à aujourd'hui + horizon. Efface l'invalidation une fois terminé. +// Retourne (zero, zero, nil) si aucune invalidation n'est en attente. +func (e *Engine) RecomputeAccount(ctx context.Context, accountID int32) (from, to time.Time, err error) { + inv, found, err := e.store.GetInvalidation(ctx, accountID) + if err != nil { + return time.Time{}, time.Time{}, fmt.Errorf("get invalidation: %w", err) + } + if !found { + return time.Time{}, time.Time{}, nil + } + + today := time.Now().Truncate(24 * time.Hour) + horizon := today.AddDate(0, 0, e.horizonDays) + + if err := e.backfillAccount(ctx, accountID, inv.RecomputeFrom, horizon); err != nil { + return time.Time{}, time.Time{}, err + } + + if err := e.store.ClearInvalidation(ctx, accountID); err != nil { + return time.Time{}, time.Time{}, fmt.Errorf("clear invalidation: %w", err) + } + + return inv.RecomputeFrom, horizon, nil +} + +// DailySnapshot : +// 1. Traite les invalidations en attente (jusqu'à today + horizon) +// 2. Recalcule hier (consolide les transactions de la veille) +// 3. Calcule le nouveau jour entrant dans la fenêtre (today + horizonDays) +func (e *Engine) DailySnapshot(ctx context.Context) error { + today := time.Now().Truncate(24 * time.Hour) + yesterday := today.AddDate(0, 0, -1) + newHorizonDay := today.AddDate(0, 0, e.horizonDays) + + // 1. Invalidations + if err := e.processInvalidations(ctx, newHorizonDay); err != nil { + e.logger.Error("snapshot: invalidation processing failed", "error", err) + } + + // 2. Hier + if err := e.RecomputeDay(ctx, yesterday); err != nil { + e.logger.Error("snapshot: yesterday recompute failed", "error", err) + } + + // 3. Nouveau jour entrant dans la fenêtre + if err := e.RecomputeDay(ctx, newHorizonDay); err != nil { + e.logger.Error("snapshot: horizon day failed", "error", err) + } + + return nil +} + +func (e *Engine) processInvalidations(ctx context.Context, until time.Time) error { + invalidations, err := e.store.GetInvalidations(ctx) + if err != nil { + return fmt.Errorf("get invalidations: %w", err) + } + if len(invalidations) == 0 { + return nil + } + + e.logger.Info("snapshot: processing invalidations", "count", len(invalidations)) + + for _, inv := range invalidations { + if err := e.backfillAccount(ctx, inv.AccountID, inv.RecomputeFrom, until); err != nil { + e.logger.Error("snapshot: backfill failed", "account_id", inv.AccountID, "error", err) + continue + } + if err := e.store.ClearInvalidation(ctx, inv.AccountID); err != nil { + e.logger.Error("snapshot: clear invalidation failed", "account_id", inv.AccountID, "error", err) + } + } + + return nil +} + +func (e *Engine) backfillAccount(ctx context.Context, accountID int32, from, to time.Time) error { + from = from.Truncate(24 * time.Hour) + to = to.Truncate(24 * time.Hour) + + if from.After(to) { + return fmt.Errorf("from (%s) after to (%s)", from.Format("2006-01-02"), to.Format("2006-01-02")) + } + + for d := from; !d.After(to); d = d.AddDate(0, 0, 1) { + if err := e.recomputeAccount(ctx, accountID, d); err != nil { + return fmt.Errorf("recompute %s: %w", d.Format("2006-01-02"), err) + } + } + return nil +} + +func (e *Engine) recomputeAccount(ctx context.Context, accountID int32, date time.Time) error { + baseDate, hasBase, err := e.store.GetLatestSnapshotDate(ctx, accountID, date) + if err != nil { + return fmt.Errorf("get latest snapshot date: %w", err) + } + + positions := map[int32]store.PositionRow{} + if hasBase { + base, err := e.store.GetBasePositions(ctx, accountID, baseDate) + if err != nil { + return fmt.Errorf("get base positions: %w", err) + } + for _, p := range base { + positions[p.InstrumentID] = p + } + } + + deltas, err := e.store.ComputeDeltas(ctx, accountID, baseDate, date) + if err != nil { + return fmt.Errorf("compute deltas: %w", err) + } + + for _, d := range deltas { + p := positions[d.InstrumentID] + p.InstrumentID = d.InstrumentID + p.InstrumentType = d.InstrumentType + p.Quantite += d.Quantite + positions[d.InstrumentID] = p + } + + var totalValeur float64 + + for _, pos := range positions { + if pos.Quantite <= 0 { + continue + } + + prix, found, err := e.store.GetPriceAt(ctx, pos.InstrumentID, date) + if err != nil { + return fmt.Errorf("get price instrument %d: %w", pos.InstrumentID, err) + } + if !found { + if pos.InstrumentType == "devise" { + prix = 1.0 + } else { + e.logger.Warn("snapshot: no price, position skipped", + "account_id", accountID, + "instrument_id", pos.InstrumentID, + "date", date.Format("2006-01-02"), + ) + continue + } + } + + valeur := pos.Quantite * prix + totalValeur += valeur + + snap := store.PositionSnapshot{ + Date: date, + AccountID: accountID, + InstrumentID: pos.InstrumentID, + Quantite: pos.Quantite, + PrixCloture: &prix, + Valeur: &valeur, + } + if err := e.store.UpsertPositionSnapshot(ctx, snap); err != nil { + return fmt.Errorf("upsert position snapshot: %w", err) + } + } + + return e.store.UpsertAccountSnapshot(ctx, store.AccountSnapshot{ + Date: date, + AccountID: accountID, + Valeur: totalValeur, + }) +} diff --git a/internal/store/snapshot.go b/internal/store/snapshot.go new file mode 100644 index 0000000..25d6d23 --- /dev/null +++ b/internal/store/snapshot.go @@ -0,0 +1,271 @@ +package store + +import ( + "context" + "errors" + "time" + + "github.com/jackc/pgx/v5" +) + +type PositionRow struct { + InstrumentID int32 `json:"instrument_id"` + InstrumentType string `json:"instrument_type"` + Quantite float64 `json:"quantite"` +} + +type PositionSnapshot struct { + Date time.Time + AccountID int32 + InstrumentID int32 + Quantite float64 + PRU *float64 + PrixCloture *float64 + Valeur *float64 +} + +type AccountSnapshot struct { + Date time.Time + AccountID int32 + Valeur float64 +} + +// ListAllAccountIDs retourne tous les comptes (maîtres + enveloppes). +func (s *Store) ListAllAccountIDs(ctx context.Context) ([]int32, error) { + rows, err := s.pool.Query(ctx, `SELECT id FROM account ORDER BY id`) + if err != nil { + return nil, err + } + defer rows.Close() + var ids []int32 + for rows.Next() { + var id int32 + if err := rows.Scan(&id); err != nil { + return nil, err + } + ids = append(ids, id) + } + return ids, rows.Err() +} + +// GetLatestSnapshotDate retourne la date du snapshot le plus récent strictement +// antérieur à before. Retourne (zero, false, nil) si aucun snapshot n'existe. +// On utilise account_snapshot comme marqueur de "jour déjà calculé" car il +// contient toujours une ligne même pour un compte à solde nul. +func (s *Store) GetLatestSnapshotDate(ctx context.Context, accountID int32, before time.Time) (time.Time, bool, error) { + var t *time.Time + err := s.pool.QueryRow(ctx, ` + SELECT MAX(date) FROM account_snapshot + WHERE account_id = $1 AND date < $2::date + `, accountID, before).Scan(&t) + if err != nil || t == nil { + return time.Time{}, false, err + } + return *t, true, nil +} + +// GetBasePositions retourne les positions d'un compte à une date de snapshot +// existante, enrichies du type d'instrument pour le fallback de prix. +func (s *Store) GetBasePositions(ctx context.Context, accountID int32, date time.Time) ([]PositionRow, error) { + rows, err := s.pool.Query(ctx, ` + SELECT ps.instrument_id, i.type AS instrument_type, ps.quantite + FROM position_snapshot ps + JOIN instrument i ON i.id = ps.instrument_id + WHERE ps.account_id = $1 AND ps.date = $2::date + `, accountID, date) + if err != nil { + return nil, err + } + return pgx.CollectRows(rows, pgx.RowToStructByName[PositionRow]) +} + +// ComputeDeltas calcule les variations nettes par instrument pour un compte +// entre fromExclusive (exclu) et toInclusive (inclus). +// Toutes les transactions sont comptées quelle que soit leur validation : +// le solde reflète l'état envisagé complet (passé confirmé + prévisionnel). +// Passer time.Time{} comme fromExclusive couvre toutes les transactions. +func (s *Store) ComputeDeltas(ctx context.Context, accountID int32, fromExclusive, toInclusive time.Time) ([]PositionRow, error) { + from := fromExclusive + if from.IsZero() { + from = time.Date(1, 1, 1, 0, 0, 0, 0, time.UTC) + } + + rows, err := s.pool.Query(ctx, ` + WITH flows AS ( + SELECT instrument_dest_id AS instrument_id, quantite_dest AS delta + FROM transaction + WHERE account_dest_id = $1 + AND date > $2::date AND date <= $3::date + UNION ALL + SELECT instrument_source_id, -quantite_source + FROM transaction + WHERE account_source_id = $1 + AND date > $2::date AND date <= $3::date + ) + SELECT f.instrument_id, i.type AS instrument_type, SUM(f.delta) AS quantite + FROM flows f + JOIN instrument i ON i.id = f.instrument_id + GROUP BY f.instrument_id, i.type + `, accountID, from, toInclusive) + if err != nil { + return nil, err + } + return pgx.CollectRows(rows, pgx.RowToStructByName[PositionRow]) +} + +// AccountSnapshotRow représente une ligne de account_snapshot pour la restitution. +type AccountSnapshotRow struct { + Date string `json:"date"` + Valeur float64 `json:"valeur"` +} + +// PositionSnapshotRow représente une ligne de position_snapshot pour la restitution. +type PositionSnapshotRow struct { + Date string `json:"date"` + InstrumentID int32 `json:"instrument_id"` + Quantite float64 `json:"quantite"` + PRU *float64 `json:"pru,omitempty"` + PrixCloture *float64 `json:"prix_cloture,omitempty"` + Valeur *float64 `json:"valeur,omitempty"` +} + +// ListAccountSnapshots retourne les snapshots agrégés d'un compte entre from et to. +// Si from est zero, pas de borne inférieure. Si to est zero, pas de borne supérieure. +func (s *Store) ListAccountSnapshots(ctx context.Context, accountID int32, from, to time.Time) ([]AccountSnapshotRow, error) { + rows, err := s.pool.Query(ctx, ` + SELECT date::text, valeur + FROM account_snapshot + WHERE account_id = $1 + AND ($2::date IS NULL OR date >= $2::date) + AND ($3::date IS NULL OR date <= $3::date) + ORDER BY date + `, accountID, nullableDate(from), nullableDate(to)) + if err != nil { + return nil, err + } + return pgx.CollectRows(rows, pgx.RowToStructByName[AccountSnapshotRow]) +} + +// ListPositionSnapshots retourne les positions détaillées d'un compte entre from et to. +// Si from est zero, pas de borne inférieure. Si to est zero, pas de borne supérieure. +func (s *Store) ListPositionSnapshots(ctx context.Context, accountID int32, from, to time.Time) ([]PositionSnapshotRow, error) { + rows, err := s.pool.Query(ctx, ` + SELECT date::text, instrument_id, quantite, pru, prix_cloture, valeur + FROM position_snapshot + WHERE account_id = $1 + AND ($2::date IS NULL OR date >= $2::date) + AND ($3::date IS NULL OR date <= $3::date) + ORDER BY date, instrument_id + `, accountID, nullableDate(from), nullableDate(to)) + if err != nil { + return nil, err + } + return pgx.CollectRows(rows, pgx.RowToStructByName[PositionSnapshotRow]) +} + +// nullableDate convertit time.Time{} en nil pour les paramètres SQL optionnels. +func nullableDate(t time.Time) any { + if t.IsZero() { + return nil + } + return t +} + +// GetPriceAt retourne le dernier prix connu pour un instrument à une date donnée. +// Retourne (0, false, nil) si aucun prix n'est trouvé. +func (s *Store) GetPriceAt(ctx context.Context, instrumentID int32, date time.Time) (float64, bool, error) { + var prix float64 + err := s.pool.QueryRow(ctx, ` + SELECT prix FROM price_history + WHERE instrument_id = $1 AND fetched_at::date <= $2::date + ORDER BY fetched_at DESC + LIMIT 1 + `, instrumentID, date).Scan(&prix) + if errors.Is(err, pgx.ErrNoRows) { + return 0, false, nil + } + if err != nil { + return 0, false, err + } + return prix, true, nil +} + +func (s *Store) UpsertPositionSnapshot(ctx context.Context, snap PositionSnapshot) error { + _, err := s.pool.Exec(ctx, ` + INSERT INTO position_snapshot (date, account_id, instrument_id, quantite, pru, prix_cloture, valeur) + VALUES ($1::date, $2, $3, $4, $5, $6, $7) + ON CONFLICT (date, account_id, instrument_id) DO UPDATE SET + quantite = EXCLUDED.quantite, + pru = EXCLUDED.pru, + prix_cloture = EXCLUDED.prix_cloture, + valeur = EXCLUDED.valeur + `, snap.Date, snap.AccountID, snap.InstrumentID, + snap.Quantite, snap.PRU, snap.PrixCloture, snap.Valeur) + return err +} + +func (s *Store) UpsertAccountSnapshot(ctx context.Context, snap AccountSnapshot) error { + _, err := s.pool.Exec(ctx, ` + INSERT INTO account_snapshot (date, account_id, valeur) + VALUES ($1::date, $2, $3) + ON CONFLICT (date, account_id) DO UPDATE SET valeur = EXCLUDED.valeur + `, snap.Date, snap.AccountID, snap.Valeur) + return err +} + +// ── Invalidation ───────────────────────────────────────────────────────────── + +type Invalidation struct { + AccountID int32 + RecomputeFrom time.Time +} + +// InvalidateSnapshot marque un compte comme nécessitant un recalcul depuis date. +// Conserve le MIN si une invalidation antérieure existe déjà. +func (s *Store) InvalidateSnapshot(ctx context.Context, accountID int32, from time.Time) error { + _, err := s.pool.Exec(ctx, ` + INSERT INTO account_snapshot_invalidation (account_id, recompute_from) + VALUES ($1, $2::date) + ON CONFLICT (account_id) DO UPDATE + SET recompute_from = LEAST(account_snapshot_invalidation.recompute_from, EXCLUDED.recompute_from) + `, accountID, from) + return err +} + +// GetInvalidations retourne tous les comptes en attente de recalcul. +func (s *Store) GetInvalidations(ctx context.Context) ([]Invalidation, error) { + rows, err := s.pool.Query(ctx, + `SELECT account_id, recompute_from FROM account_snapshot_invalidation ORDER BY account_id`) + if err != nil { + return nil, err + } + defer rows.Close() + var out []Invalidation + for rows.Next() { + var inv Invalidation + if err := rows.Scan(&inv.AccountID, &inv.RecomputeFrom); err != nil { + return nil, err + } + out = append(out, inv) + } + return out, rows.Err() +} + +// ClearInvalidation supprime le flag d'invalidation une fois le recalcul terminé. +func (s *Store) ClearInvalidation(ctx context.Context, accountID int32) error { + _, err := s.pool.Exec(ctx, + `DELETE FROM account_snapshot_invalidation WHERE account_id = $1`, accountID) + return err +} + +// GetInvalidation retourne l'invalidation d'un compte spécifique, si elle existe. +func (s *Store) GetInvalidation(ctx context.Context, accountID int32) (Invalidation, bool, error) { + var inv Invalidation + err := s.pool.QueryRow(ctx, + `SELECT account_id, recompute_from FROM account_snapshot_invalidation WHERE account_id = $1`, + accountID).Scan(&inv.AccountID, &inv.RecomputeFrom) + if errors.Is(err, pgx.ErrNoRows) { + return Invalidation{}, false, nil + } + return inv, err == nil, err +} diff --git a/internal/store/transaction.go b/internal/store/transaction.go index 83f5836..f5da808 100644 --- a/internal/store/transaction.go +++ b/internal/store/transaction.go @@ -3,6 +3,7 @@ package store import ( "context" "fmt" + "time" "github.com/jackc/pgx/v5" ) @@ -129,10 +130,21 @@ func (s *Store) CreateTransaction(ctx context.Context, p CreateTransactionParams if err != nil { return Transaction{}, err } - return pgx.CollectOneRow(rows, pgx.RowToStructByName[Transaction]) + tx, err := pgx.CollectOneRow(rows, pgx.RowToStructByName[Transaction]) + if err != nil { + return tx, err + } + s.invalidateAccounts(ctx, tx.Date, tx.AccountSourceID, tx.AccountDestID) + return tx, nil } func (s *Store) UpdateTransaction(ctx context.Context, id int64, p CreateTransactionParams) (Transaction, error) { + // Récupérer l'ancienne date pour invalider à partir du MIN(ancienne, nouvelle). + old, err := s.GetTransaction(ctx, id) + if err != nil { + return Transaction{}, err + } + rows, err := s.pool.Query(ctx, `UPDATE transaction SET date = $2::date, @@ -148,7 +160,16 @@ func (s *Store) UpdateTransaction(ctx context.Context, id int64, p CreateTransac if err != nil { return Transaction{}, err } - return pgx.CollectOneRow(rows, pgx.RowToStructByName[Transaction]) + tx, err := pgx.CollectOneRow(rows, pgx.RowToStructByName[Transaction]) + if err != nil { + return tx, err + } + // Invalider depuis la date la plus ancienne (ancienne ou nouvelle). + earliest := minDateStr(old.Date, tx.Date) + s.invalidateAccounts(ctx, earliest, tx.AccountSourceID, tx.AccountDestID) + // Si les comptes ont changé, invalider aussi les anciens. + s.invalidateAccounts(ctx, earliest, old.AccountSourceID, old.AccountDestID) + return tx, nil } func (s *Store) SetValidated(ctx context.Context, id int64, validated bool) (Transaction, error) { @@ -158,10 +179,45 @@ func (s *Store) SetValidated(ctx context.Context, id int64, validated bool) (Tra if err != nil { return Transaction{}, err } - return pgx.CollectOneRow(rows, pgx.RowToStructByName[Transaction]) + tx, err := pgx.CollectOneRow(rows, pgx.RowToStructByName[Transaction]) + if err != nil { + return tx, err + } + s.invalidateAccounts(ctx, tx.Date, tx.AccountSourceID, tx.AccountDestID) + return tx, nil } func (s *Store) DeleteTransaction(ctx context.Context, id int64) error { - _, err := s.pool.Exec(ctx, `DELETE FROM transaction WHERE id = $1`, id) - return err + tx, err := s.GetTransaction(ctx, id) + if err != nil { + return err + } + if _, err := s.pool.Exec(ctx, `DELETE FROM transaction WHERE id = $1`, id); err != nil { + return err + } + s.invalidateAccounts(ctx, tx.Date, tx.AccountSourceID, tx.AccountDestID) + return nil +} + +// invalidateAccounts marque les comptes non-nil comme devant être recalculés depuis dateStr. +func (s *Store) invalidateAccounts(ctx context.Context, dateStr string, accountIDs ...*int32) { + date, err := time.Parse("2006-01-02", dateStr) + if err != nil { + return + } + seen := map[int32]bool{} + for _, id := range accountIDs { + if id != nil && !seen[*id] { + seen[*id] = true + s.InvalidateSnapshot(ctx, *id, date) //nolint:errcheck + } + } +} + +// minDateStr retourne la plus petite des deux dates au format YYYY-MM-DD. +func minDateStr(a, b string) string { + if a <= b { + return a + } + return b } diff --git a/migrations/000002_snapshot_invalidation.down.sql b/migrations/000002_snapshot_invalidation.down.sql new file mode 100644 index 0000000..eeae05c --- /dev/null +++ b/migrations/000002_snapshot_invalidation.down.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS account_snapshot_invalidation; diff --git a/migrations/000002_snapshot_invalidation.up.sql b/migrations/000002_snapshot_invalidation.up.sql new file mode 100644 index 0000000..7e08b4a --- /dev/null +++ b/migrations/000002_snapshot_invalidation.up.sql @@ -0,0 +1,7 @@ +-- Marque les comptes dont les snapshots doivent être recalculés. +-- recompute_from = date la plus ancienne à partir de laquelle recalculer. +-- Mise à jour par MIN() à chaque mutation de transaction (create/update/delete/validate). +CREATE TABLE account_snapshot_invalidation ( + account_id INTEGER PRIMARY KEY REFERENCES account(id) ON DELETE CASCADE, + recompute_from DATE NOT NULL +);