tiennm99/miti99bot
1
0
Fork 0
mirror of https://github.com/tiennm99/miti99bot.git synced 2026-09-04 08:19:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(stock): persist per-user dividend history

Browse Source
This commit is contained in:
tiennm99
2026-07-22 16:17:29 +07:00
parent b6f6361c94
commit fda0bbd0b1
25 changed files with 2076 additions and 582 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/modules/stock/dividend_callback.go
+46 -42
View File
@@ -32,6 +32,19 @@ func removeDividendButton(ctx context.Context, b *bot.Bot, chatID int64, message
return err
}
func (s *state) invalidateDividendAction(ctx context.Context, b *bot.Bot, actionKey string, action PendingDividendAction) (error, error) {
var deleteErr error
if s.pending != nil {
deleteErr = s.pending.Delete(ctx, actionKey)
}
return deleteErr, removeDividendButton(ctx, b, action.ChatID, action.MessageID)
}
func (s *state) rejectDividendAction(ctx context.Context, b *bot.Bot, queryID, actionKey string, action PendingDividendAction, text string) error {
_, _ = s.invalidateDividendAction(ctx, b, actionKey, action)
return answerDividendCallback(ctx, b, queryID, text, true)
}
func (s *state) handleDividendCallback(ctx context.Context, b *bot.Bot, update *models.Update) error {
if update == nil || update.CallbackQuery == nil {
return nil
@@ -73,9 +86,7 @@ func (s *state) resolveDividendCallback(ctx context.Context, b *bot.Bot, query *
}
now := s.now().UnixMilli()
if action.ExpiresAt <= now {
_ = s.pending.Delete(ctx, actionKey)
_ = removeDividendButton(ctx, b, action.ChatID, action.MessageID)
err = answerDividendCallback(ctx, b, query.ID, "This suggestion expired. Run /stock_portfolio again.", true)
err = s.rejectDividendAction(ctx, b, query.ID, actionKey, action, "This suggestion expired. Run /stock_portfolio again.")
return PendingDividendAction{}, nil, "", true, err
}
return action, msg, actionKey, false, nil
@@ -97,45 +108,47 @@ func (s *state) consumeDividendAction(ctx context.Context, b *bot.Bot, query *mo
log.Error("stock_load_portfolio", "user", action.OwnerUserID, "err", err)
return answerDividendCallback(ctx, b, query.ID, "Could not load your portfolio. Try again.", true)
}
eventKey := dividendLedgerKey(action.ProviderEventID)
if _, applied := p.AppliedDividendEvents[eventKey]; applied {
_ = s.pending.Delete(ctx, actionKey)
_ = removeDividendButton(ctx, b, action.ChatID, action.MessageID)
return answerDividendCallback(ctx, b, query.ID, "This dividend was already applied.", true)
record, exists := p.dividendRecord(action.Symbol, action.ProviderEventID)
if !exists {
return s.rejectDividendAction(ctx, b, query.ID, actionKey, action, "This dividend is no longer available.")
}
if record.Processed {
return s.rejectDividendAction(ctx, b, query.ID, actionKey, action, "This dividend was already applied.")
}
if !dividendRecordDue(record, s.now()) {
return s.rejectDividendAction(ctx, b, query.ID, actionKey, action, "This dividend is not available before Record date.")
}
position, held := p.Assets[action.Symbol]
if !held || position.Quantity <= 0 {
_ = s.pending.Delete(ctx, actionKey)
_ = removeDividendButton(ctx, b, action.ChatID, action.MessageID)
return answerDividendCallback(ctx, b, query.ID, "You no longer hold "+action.Symbol+". Dividend not applied.", true)
return s.rejectDividendAction(ctx, b, query.ID, actionKey, action, "You no longer hold "+action.Symbol+". Dividend not applied.")
}
if position.OpenedAt != action.PositionOpenedAt {
_ = s.pending.Delete(ctx, actionKey)
_ = removeDividendButton(ctx, b, action.ChatID, action.MessageID)
return answerDividendCallback(ctx, b, query.ID, "This "+action.Symbol+" position was closed after the suggestion. Run /stock_portfolio again.", true)
return s.rejectDividendAction(ctx, b, query.ID, actionKey, action, "This "+action.Symbol+" position was closed after the suggestion. Run /stock_portfolio again.")
}
if !positionOpenedByRecordDate(position, record) {
return s.rejectDividendAction(ctx, b, query.ID, actionKey, action, "This "+action.Symbol+" position was opened after Record date. Dividend not applied.")
}
result, err := applySuggestedDividend(&p, action, position.Quantity, now)
result, err := applySuggestedDividend(&p, action.Symbol, record, position.Quantity, now)
if err != nil {
log.Error("stock_apply_suggested_dividend", "user", action.OwnerUserID, "ticker", action.Symbol, "event", action.ProviderEventID, "err", err)
return answerDividendCallback(ctx, b, query.ID, "Could not safely apply this dividend.", true)
}
if p.AppliedDividendEvents == nil {
p.AppliedDividendEvents = map[string]int64{}
}
p.AppliedDividendEvents[eventKey] = now
record.Processed = true
p.setDividendRecord(action.Symbol, action.ProviderEventID, record)
if err := SavePortfolio(ctx, s.store, action.OwnerUserID, p); err != nil {
log.Error("stock_save_portfolio", "user", action.OwnerUserID, "err", err)
return answerDividendCallback(ctx, b, query.ID, "Could not save your portfolio. Try again.", true)
}
// The portfolio ledger is the source of truth for idempotency. Cleanup and
// The portfolio history is the source of truth for idempotency. Cleanup and
// Telegram UI updates are best-effort after the atomic portfolio save.
if err := s.pending.Delete(ctx, actionKey); err != nil {
log.Error("stock_delete_dividend_action", "err", err)
deleteErr, removeErr := s.invalidateDividendAction(ctx, b, actionKey, action)
if deleteErr != nil {
log.Error("stock_delete_dividend_action", "err", deleteErr)
}
if err := removeDividendButton(ctx, b, action.ChatID, action.MessageID); err != nil {
log.Error("stock_remove_dividend_button", "user", action.OwnerUserID, "ticker", action.Symbol, "err", err)
if removeErr != nil {
log.Error("stock_remove_dividend_button", "user", action.OwnerUserID, "ticker", action.Symbol, "err", removeErr)
}
if err := answerDividendCallback(ctx, b, query.ID, "Dividend applied.", false); err != nil {
log.Error("stock_answer_dividend_callback", "user", action.OwnerUserID, "ticker", action.Symbol, "err", err)
@@ -143,11 +156,10 @@ func (s *state) consumeDividendAction(ctx context.Context, b *bot.Bot, query *mo
return chathelper.Reply(ctx, b, msg, result)
}
func applySuggestedDividend(p *Portfolio, action PendingDividendAction, held, now int64) (string, error) {
preservedCursor := max(p.Assets[action.Symbol].DividendCheckedAt, action.CheckThrough)
switch action.Kind {
func applySuggestedDividend(p *Portfolio, symbol string, record DividendRecord, held, now int64) (string, error) {
switch record.Kind {
case DividendKindCash:
total, err := cashDividendTotal(held, action.VNDPerShare)
total, err := cashDividendTotal(held, record.VNDPerShare)
if err != nil {
return "", err
}
@@ -155,17 +167,16 @@ func applySuggestedDividend(p *Portfolio, action PendingDividendAction, held, no
if err != nil {
return "", err
}
if err := p.ApplyDividend(action.Symbol, held, balance, now); err != nil {
if err := p.ApplyDividend(symbol, held, balance, now); err != nil {
return "", err
}
preserveDividendCursor(p, action.Symbol, preservedCursor)
return "Applied cash dividend for " + action.Symbol + ": " + FormatVND(float64(action.VNDPerShare)) +
return "Applied cash dividend for " + symbol + ": " + FormatVND(float64(record.VNDPerShare)) +
" × " + formatShareQuantity(held) + " = " + FormatVND(float64(total)) +
"\nBalance: " + FormatVND(balance), nil
case DividendKindShares:
ratio := shareRatio{owned: action.OwnedShares, new: action.NewShares,
raw: strconv.FormatInt(action.OwnedShares, 10) + ":" + strconv.FormatInt(action.NewShares, 10)}
ratio := shareRatio{owned: record.OwnedShares, new: record.NewShares,
raw: strconv.FormatInt(record.OwnedShares, 10) + ":" + strconv.FormatInt(record.NewShares, 10)}
newShares, err := shareDividendEntitlement(held, ratio)
if err != nil {
return "", err
@@ -177,20 +188,13 @@ func applySuggestedDividend(p *Portfolio, action PendingDividendAction, held, no
if err != nil {
return "", err
}
if err := p.ApplyDividend(action.Symbol, finalHolding, p.VND, now); err != nil {
if err := p.ApplyDividend(symbol, finalHolding, p.VND, now); err != nil {
return "", err
}
preserveDividendCursor(p, action.Symbol, preservedCursor)
return "Applied share dividend for " + action.Symbol + " (" + ratio.raw + "): +" +
return "Applied share dividend for " + symbol + " (" + ratio.raw + "): +" +
formatShareQuantity(newShares) + "\nHolding: " + formatShareQuantity(held) + " → " +
formatShareQuantity(finalHolding), nil
default:
return "", errors.New("unsupported dividend kind")
}
}
func preserveDividendCursor(p *Portfolio, symbol string, cursor int64) {
position := p.Assets[symbol]
position.DividendCheckedAt = cursor
p.Assets[symbol] = position
}
internal/modules/stock/dividend_flow_test.go
+367 -276
View File
@@ -1,11 +1,8 @@
package stock
import (
"bytes"
"context"
"encoding/json"
"errors"
"log/slog"
"net/http"
"net/http/httptest"
"strings"
@@ -15,72 +12,36 @@ import (
"github.com/go-telegram/bot/models"
applog "github.com/tiennm99/miti99bot/internal/log"
"github.com/tiennm99/miti99bot/internal/storage"
"github.com/tiennm99/miti99bot/internal/testutil"
)
type dividendProviderCall struct {
symbol string
after, through time.Time
}
type fakeDividendProvider struct {
mu sync.Mutex
events []DividendEvent
err error
calls int
calls []dividendProviderCall
}
type blockingDividendProvider struct {
started chan struct{}
release chan struct{}
events []DividendEvent
}
func (p *blockingDividendProvider) FetchDividendEvents(_ context.Context, _ string, _, _ time.Time) ([]DividendEvent, error) {
close(p.started)
<-p.release
return append([]DividendEvent(nil), p.events...), nil
}
func (f *fakeDividendProvider) FetchDividendEvents(_ context.Context, _ string, _, _ time.Time) ([]DividendEvent, error) {
f.calls++
func (f *fakeDividendProvider) FetchDividendEvents(_ context.Context, symbol string, after, through time.Time) ([]DividendEvent, error) {
f.mu.Lock()
defer f.mu.Unlock()
f.calls = append(f.calls, dividendProviderCall{symbol: symbol, after: after, through: through})
return append([]DividendEvent(nil), f.events...), f.err
}
func TestPortfolioDividendCheckLogsNoEvents(t *testing.T) {
s, store, _, rb, now := newDividendFlowState(t, nil)
seedDividendFlowPortfolio(t, store, now, 100)
p, err := LoadPortfolio(context.Background(), store, 7, now.UnixMilli())
if err != nil {
t.Fatal(err)
}
var output bytes.Buffer
previous := applog.Default()
applog.SetDefault(slog.New(slog.NewJSONHandler(&output, nil)))
t.Cleanup(func() { applog.SetDefault(previous) })
if err := s.notifyDividendEvents(
context.Background(),
rb.Bot,
testutil.NewPrivateMessage(7, "/stock_portfolio").Message,
7,
p,
now,
); err != nil {
t.Fatal(err)
}
var record map[string]any
if err := json.Unmarshal(bytes.TrimSpace(output.Bytes()), &record); err != nil {
t.Fatalf("decode dividend check log %q: %v", output.String(), err)
}
if record["msg"] != "stock_dividend_events_checked" ||
record["user"] != float64(7) ||
record["ticker"] != "TCB" ||
record["events"] != float64(0) ||
record["status"] != "success" {
t.Fatalf("unexpected dividend check log: %#v", record)
}
func (f *fakeDividendProvider) snapshotCalls() []dividendProviderCall {
f.mu.Lock()
defer f.mu.Unlock()
return append([]dividendProviderCall(nil), f.calls...)
}
func newDividendFlowState(t *testing.T, events []DividendEvent) (*state, Store, PendingDividendStore, *testutil.RecordingBot, time.Time) {
func newDividendFlowState(t *testing.T, events []DividendEvent) (*state, Store, PendingDividendStore, *testutil.RecordingBot, time.Time, *fakeDividendProvider) {
t.Helper()
now := time.Date(2026, 6, 25, 12, 0, 0, 0, saigonLocation)
priceServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
@@ -93,18 +54,19 @@ func newDividendFlowState(t *testing.T, events []DividendEvent) (*state, Store,
collection := provider.Collection(CollectionName)
store := storage.Typed[Portfolio](collection)
pending := storage.Typed[PendingDividendAction](collection)
fake := &fakeDividendProvider{events: events}
tokenIndex := 0
s := &state{
store: store,
pending: pending,
store: store, pending: pending,
prices: &PriceClient{HTTP: priceServer.Client(), URL: priceServer.URL},
dividends: &fakeDividendProvider{events: events},
nowFn: func() time.Time { return now },
dividends: fake, nowFn: func() time.Time { return now },
newDividendToken: func() (string, error) {
return "abcdefghijklmnopqrstuv", nil
token := "abcdefghijklmnopqrstu" + string(rune('v'+tokenIndex))
tokenIndex++
return token, nil
},
}
rb := testutil.NewRecordingBot(t)
return s, store, pending, rb, now
return s, store, pending, testutil.NewRecordingBot(t), now, fake
}
func seedDividendFlowPortfolio(t *testing.T, store Store, now time.Time, quantity int64) {
@@ -120,242 +82,365 @@ func seedDividendFlowPortfolio(t *testing.T, store Store, now time.Time, quantit
}
}
func TestStockPortfolio_NoDividendEventsSendsOnlyPortfolio(t *testing.T) {
s, store, _, rb, now := newDividendFlowState(t, nil)
seedDividendFlowPortfolio(t, store, now, 100)
if err := s.handleStats(context.Background(), rb.Bot, testutil.NewPrivateMessage(7, "/stock_portfolio")); err != nil {
t.Fatal(err)
}
calls := rb.Sent()
if len(calls) != 1 || !strings.Contains(calls[0].Text(), "Stock Portfolio") {
t.Fatalf("calls = %#v, want only portfolio message", calls)
}
p, err := LoadPortfolio(context.Background(), store, 7, now.UnixMilli())
if err != nil {
t.Fatal(err)
}
if got := p.Assets["TCB"].DividendCheckedAt; got != now.UnixMilli() {
t.Fatalf("dividend cursor = %d, want %d", got, now.UnixMilli())
}
}
func TestStockPortfolio_DividendEventFollowsPortfolioWithOpaqueButton(t *testing.T) {
event := DividendEvent{
func cashDividendEvent(now, recordDate time.Time) DividendEvent {
return DividendEvent{
ProviderID: "2612974", Symbol: "TCB", Kind: DividendKindCash,
PublishedAt: time.Date(2026, 6, 24, 8, 0, 0, 0, saigonLocation),
ExDate: time.Date(2026, 6, 27, 0, 0, 0, 0, saigonLocation),
PublishedAt: now.Add(-24 * time.Hour), RecordDate: recordDate,
ExDate: now.AddDate(0, 0, 1), PaymentDate: now.AddDate(0, 0, 10),
VNDPerShare: 1500, Title: "Cash dividend",
}
s, store, pending, rb, now := newDividendFlowState(t, []DividendEvent{event})
}
func TestStockPortfolioNoEventsUsesExactRecentWindow(t *testing.T) {
s, store, _, rb, now, provider := newDividendFlowState(t, nil)
seedDividendFlowPortfolio(t, store, now, 100)
if err := s.handleStats(context.Background(), rb.Bot, testutil.NewPrivateMessage(7, "/stock_portfolio")); err != nil {
t.Fatal(err)
}
if calls := rb.Sent(); len(calls) != 1 || !strings.Contains(calls[0].Text(), "Stock Portfolio") {
t.Fatalf("messages = %#v", calls)
}
providerCalls := provider.snapshotCalls()
if len(providerCalls) != 1 || !providerCalls[0].after.Equal(now.Add(-30*24*time.Hour)) || !providerCalls[0].through.Equal(now) {
t.Fatalf("provider calls = %+v", providerCalls)
}
}
func TestFutureDividendNotifiesOnEveryPortfolioWithoutButton(t *testing.T) {
event := cashDividendEvent(time.Date(2026, 6, 25, 12, 0, 0, 0, saigonLocation), time.Date(2026, 6, 27, 0, 0, 0, 0, saigonLocation))
s, store, pending, rb, now, _ := newDividendFlowState(t, []DividendEvent{event})
seedDividendFlowPortfolio(t, store, now, 100)
for range 2 {
if err := s.handleStats(context.Background(), rb.Bot, testutil.NewPrivateMessage(7, "/stock_portfolio")); err != nil {
t.Fatal(err)
}
}
calls := rb.Sent()
if len(calls) != 4 || !strings.Contains(calls[1].Text(), "Upcoming dividend event") || !strings.Contains(calls[3].Text(), "Upcoming dividend event") || calls[1].Form["reply_markup"] != "" || calls[3].Form["reply_markup"] != "" {
t.Fatalf("messages = %#v", calls)
}
keys, err := pending.List(context.Background(), pendingDividendPrefix)
if err != nil || len(keys) != 0 {
t.Fatalf("pending = %v, err=%v", keys, err)
}
p, _ := LoadPortfolio(context.Background(), store, 7, now.UnixMilli())
record := p.Dividends["TCB"][event.ProviderID]
if record.Processed || record.VNDPerShare != 1500 {
t.Fatalf("stored event = %+v", record)
}
}
func TestMissingRecordDateRemainsInformational(t *testing.T) {
now := time.Date(2026, 6, 25, 12, 0, 0, 0, saigonLocation)
event := cashDividendEvent(now, time.Time{})
s, store, pending, rb, _, _ := newDividendFlowState(t, []DividendEvent{event})
seedDividendFlowPortfolio(t, store, now, 100)
if err := s.handleStats(context.Background(), rb.Bot, testutil.NewPrivateMessage(7, "/stock_portfolio")); err != nil {
t.Fatal(err)
}
calls := rb.Sent()
if len(calls) != 2 || !strings.Contains(calls[0].Text(), "Stock Portfolio") || !strings.Contains(calls[1].Text(), "Recent dividend event") {
t.Fatalf("unexpected message order: %#v", calls)
}
markup := calls[1].Form["reply_markup"]
if !strings.Contains(markup, dividendCallbackPrefix+"abcdefghijklmnopqrstuv") || strings.Contains(markup, "1500") || strings.Contains(markup, "2612974") {
t.Fatalf("callback markup is not opaque: %q", markup)
if len(calls) != 2 || !strings.Contains(calls[1].Text(), "awaiting SSI update") || calls[1].Form["reply_markup"] != "" {
t.Fatalf("messages = %#v", calls)
}
keys, err := pending.List(context.Background(), pendingDividendPrefix)
if err != nil || len(keys) != 1 {
t.Fatalf("pending keys = %v, err=%v", keys, err)
}
action, _, err := pending.Get(context.Background(), keys[0])
if err != nil || action.OwnerUserID != 7 || action.ChatID != 7 || action.MessageID != 1 {
t.Fatalf("pending action = %+v, err=%v", action, err)
if err != nil || len(keys) != 0 {
t.Fatalf("pending = %v, err=%v", keys, err)
}
}
func TestDividendCallback_OnlyOwnerCanApplyAndUsesCurrentHolding(t *testing.T) {
s, store, pending, rb, now := newDividendFlowState(t, nil)
func TestMissingRecordDateIsRefetchedFromOriginalPublicationDay(t *testing.T) {
now := time.Date(2026, 6, 25, 12, 0, 0, 0, saigonLocation)
oldPublished := now.AddDate(0, 0, -45)
event := cashDividendEvent(now, startOfSaigonDay(now))
event.PublishedAt = oldPublished
s, store, pending, rb, _, provider := newDividendFlowState(t, []DividendEvent{event})
seedDividendFlowPortfolio(t, store, now, 100)
action := PendingDividendAction{
OwnerUserID: 7, ChatID: -100, MessageID: 99, ProviderEventID: "2612974",
Symbol: "TCB", Kind: DividendKindCash, VNDPerShare: 1500,
ObservedHolding: 50, PositionOpenedAt: now.Add(-48 * time.Hour).UnixMilli(), CheckThrough: now.UnixMilli(), CreatedAt: now.UnixMilli(), ExpiresAt: now.Add(time.Hour).UnixMilli(),
}
key := pendingDividendKey("abcdefghijklmnopqrstuv")
if err := pending.Put(context.Background(), key, action); err != nil {
t.Fatal(err)
}
callback := func(userID int64) *models.Update {
return &models.Update{CallbackQuery: &models.CallbackQuery{
ID: "query", From: models.User{ID: userID}, Data: dividendCallbackPrefix + "abcdefghijklmnopqrstuv",
Message: models.MaybeInaccessibleMessage{Type: models.MaybeInaccessibleMessageTypeMessage, Message: &models.Message{
ID: 99, Chat: models.Chat{ID: -100, Type: models.ChatTypeGroup}, MessageThreadID: 3,
}},
}}
}
if err := s.handleDividendCallback(context.Background(), rb.Bot, callback(8)); err != nil {
t.Fatal(err)
}
p, err := LoadPortfolio(context.Background(), store, 7, now.UnixMilli())
if err != nil || p.VND != 100_000 {
t.Fatalf("unauthorized click changed portfolio: %+v, err=%v", p, err)
}
if _, _, err := pending.Get(context.Background(), key); err != nil {
t.Fatalf("unauthorized click consumed action: %v", err)
}
s.nowFn = func() time.Time { return now.Add(30 * time.Minute) }
if err := s.handleDividendCallback(context.Background(), rb.Bot, callback(7)); err != nil {
t.Fatal(err)
}
p, err = LoadPortfolio(context.Background(), store, 7, now.UnixMilli())
if err != nil {
t.Fatal(err)
}
if p.VND != 250_000 || p.Assets["TCB"].Quantity != 100 || p.Assets["TCB"].DividendCheckedAt != now.UnixMilli() || len(p.AppliedDividendEvents) != 1 {
t.Fatalf("applied portfolio = %+v", p)
}
if err := s.handleDividendCallback(context.Background(), rb.Bot, callback(7)); err != nil {
t.Fatal(err)
}
p, _ = LoadPortfolio(context.Background(), store, 7, now.UnixMilli())
if p.VND != 250_000 {
t.Fatalf("repeated click credited twice: %v", p.VND)
}
}
func TestDividendCallback_AppliesShareDividendOnce(t *testing.T) {
s, store, pending, rb, now := newDividendFlowState(t, nil)
seedDividendFlowPortfolio(t, store, now, 139)
action := PendingDividendAction{
OwnerUserID: 7, ChatID: 7, MessageID: 1, ProviderEventID: "2612975",
Symbol: "TCB", Kind: DividendKindShares, OwnedShares: 100, NewShares: 10,
ObservedHolding: 139, PositionOpenedAt: now.Add(-48 * time.Hour).UnixMilli(), CheckThrough: now.UnixMilli(), CreatedAt: now.UnixMilli(), ExpiresAt: now.Add(time.Hour).UnixMilli(),
}
if err := pending.Put(context.Background(), pendingDividendKey("abcdefghijklmnopqrstuv"), action); err != nil {
t.Fatal(err)
}
update := &models.Update{CallbackQuery: &models.CallbackQuery{
ID: "query", From: models.User{ID: 7}, Data: dividendCallbackPrefix + "abcdefghijklmnopqrstuv",
Message: models.MaybeInaccessibleMessage{Type: models.MaybeInaccessibleMessageTypeMessage, Message: &models.Message{
ID: 1, Chat: models.Chat{ID: 7, Type: models.ChatTypePrivate},
}},
}}
if err := s.handleDividendCallback(context.Background(), rb.Bot, update); err != nil {
t.Fatal(err)
}
p, err := LoadPortfolio(context.Background(), store, 7, now.UnixMilli())
if err != nil {
t.Fatal(err)
}
if p.Assets["TCB"].Quantity != 152 || p.Assets["TCB"].Base != 139*30_000 {
t.Fatalf("share dividend result = %+v", p.Assets["TCB"])
}
}
func TestDividendCallback_RejectsSoldAndReopenedPosition(t *testing.T) {
s, store, pending, rb, now := newDividendFlowState(t, nil)
seedDividendFlowPortfolio(t, store, now, 100)
action := PendingDividendAction{
OwnerUserID: 7, ChatID: 7, MessageID: 1, ProviderEventID: "2612974",
Symbol: "TCB", Kind: DividendKindCash, VNDPerShare: 1500,
PositionOpenedAt: now.Add(-48 * time.Hour).UnixMilli(), CheckThrough: now.UnixMilli(), CreatedAt: now.UnixMilli(), ExpiresAt: now.Add(time.Hour).UnixMilli(),
}
key := pendingDividendKey("abcdefghijklmnopqrstuv")
if err := pending.Put(context.Background(), key, action); err != nil {
t.Fatal(err)
}
p, _ := LoadPortfolio(context.Background(), store, 7, now.UnixMilli())
if _, _, ok, err := p.SellTicker("TCB", 100); err != nil || !ok {
t.Fatalf("sell before reopen: ok=%v err=%v", ok, err)
}
if err := p.BuyTicker("TCB", 50, 2_000_000, now.Add(time.Minute).UnixMilli()); err != nil {
t.Fatal(err)
p.Dividends["TCB"] = map[string]DividendRecord{
event.ProviderID: {
Kind: DividendKindCash, PublishedAt: oldPublished.UnixMilli(),
VNDPerShare: 1000,
},
}
if err := SavePortfolio(context.Background(), store, 7, p); err != nil {
t.Fatal(err)
}
update := dividendCallbackUpdate(7, 7, 1)
if err := s.handleDividendCallback(context.Background(), rb.Bot, update); err != nil {
if err := s.handleStats(context.Background(), rb.Bot, testutil.NewPrivateMessage(7, "/stock_portfolio")); err != nil {
t.Fatal(err)
}
p, _ = LoadPortfolio(context.Background(), store, 7, now.UnixMilli())
if p.VND != 100_000 || p.Assets["TCB"].Quantity != 50 {
t.Fatalf("stale action changed reopened position: %+v", p)
providerCalls := provider.snapshotCalls()
foundHistoricalWindow := false
for _, call := range providerCalls {
if startOfSaigonDay(call.after).Equal(startOfSaigonDay(oldPublished)) {
foundHistoricalWindow = true
break
}
}
if _, _, err := pending.Get(context.Background(), key); !errors.Is(err, storage.ErrNotFound) {
t.Fatalf("stale action was not invalidated: %v", err)
if len(providerCalls) != 2 || !foundHistoricalWindow {
t.Fatalf("provider calls = %+v", providerCalls)
}
keys, err := pending.List(context.Background(), pendingDividendPrefix)
if err != nil || len(keys) != 1 {
t.Fatalf("pending = %v, err=%v", keys, err)
}
p, _ = LoadPortfolio(context.Background(), store, 7, now.UnixMilli())
if p.Dividends["TCB"][event.ProviderID].RecordDate == 0 || p.Dividends["TCB"][event.ProviderID].VNDPerShare != 1500 {
t.Fatalf("refreshed event = %+v", p.Dividends["TCB"][event.ProviderID])
}
}
func TestDividendCallback_ExpiredActionDoesNotApply(t *testing.T) {
s, store, pending, rb, now := newDividendFlowState(t, nil)
func TestExpiredIncompleteEventIsNotRestoredByHistoricalRefresh(t *testing.T) {
now := time.Date(2026, 6, 25, 12, 0, 0, 0, saigonLocation)
oldPublished := now.AddDate(0, 0, -90)
event := cashDividendEvent(now, time.Time{})
event.PublishedAt = oldPublished
s, store, _, rb, _, _ := newDividendFlowState(t, []DividendEvent{event})
seedDividendFlowPortfolio(t, store, now, 100)
action := PendingDividendAction{
OwnerUserID: 7, ChatID: 7, MessageID: 1, ProviderEventID: "2612974",
Symbol: "TCB", Kind: DividendKindCash, VNDPerShare: 1500,
PositionOpenedAt: now.Add(-48 * time.Hour).UnixMilli(), CheckThrough: now.UnixMilli(), CreatedAt: now.Add(-2 * time.Hour).UnixMilli(), ExpiresAt: now.Add(-time.Hour).UnixMilli(),
p, _ := LoadPortfolio(context.Background(), store, 7, now.UnixMilli())
p.Dividends["TCB"] = map[string]DividendRecord{
event.ProviderID: {Kind: DividendKindCash, PublishedAt: oldPublished.UnixMilli(), VNDPerShare: 1500},
}
key := pendingDividendKey("abcdefghijklmnopqrstuv")
if err := pending.Put(context.Background(), key, action); err != nil {
if err := SavePortfolio(context.Background(), store, 7, p); err != nil {
t.Fatal(err)
}
if err := s.handleDividendCallback(context.Background(), rb.Bot, dividendCallbackUpdate(7, 7, 1)); err != nil {
if err := s.handleStats(context.Background(), rb.Bot, testutil.NewPrivateMessage(7, "/stock_portfolio")); err != nil {
t.Fatal(err)
}
p, _ = LoadPortfolio(context.Background(), store, 7, now.UnixMilli())
if _, exists := p.Dividends["TCB"]; exists {
t.Fatalf("expired event was restored: %+v", p.Dividends)
}
}
func TestRecordDateCreatesSeparateActionableMessages(t *testing.T) {
now := time.Date(2026, 6, 25, 12, 0, 0, 0, saigonLocation)
cash := cashDividendEvent(now, startOfSaigonDay(now))
shares := DividendEvent{
ProviderID: "2612975", Symbol: "TCB", Kind: DividendKindShares,
PublishedAt: now.Add(-12 * time.Hour), RecordDate: startOfSaigonDay(now),
OwnedShares: 100, NewShares: 10, Title: "Share dividend",
}
s, store, pending, rb, _, _ := newDividendFlowState(t, []DividendEvent{cash, shares})
seedDividendFlowPortfolio(t, store, now, 139)
for range 2 {
if err := s.handleStats(context.Background(), rb.Bot, testutil.NewPrivateMessage(7, "/stock_portfolio")); err != nil {
t.Fatal(err)
}
}
calls := rb.Sent()
if len(calls) != 6 || !strings.Contains(calls[1].Text(), "Dividend ready to apply") || !strings.Contains(calls[2].Text(), "Dividend ready to apply") || !strings.Contains(calls[4].Text(), "Dividend ready to apply") || !strings.Contains(calls[5].Text(), "Dividend ready to apply") {
t.Fatalf("messages = %#v", calls)
}
keys, err := pending.List(context.Background(), pendingDividendPrefix)
if err != nil || len(keys) != 4 {
t.Fatalf("pending = %v, err=%v", keys, err)
}
}
func TestFailedFutureNoticeKeepsStoredHistoryForRetry(t *testing.T) {
now := time.Date(2026, 6, 25, 12, 0, 0, 0, saigonLocation)
event := cashDividendEvent(now, now.AddDate(0, 0, 2))
s, store, _, rb, _, _ := newDividendFlowState(t, []DividendEvent{event})
seedDividendFlowPortfolio(t, store, now, 100)
snapshot, _ := LoadPortfolio(context.Background(), store, 7, now.UnixMilli())
rb.FailMethod("sendMessage", http.StatusInternalServerError, "")
_ = s.notifyDividendEvents(context.Background(), rb.Bot, testutil.NewPrivateMessage(7, "/stock_portfolio").Message, 7, snapshot, now)
p, _ := LoadPortfolio(context.Background(), store, 7, now.UnixMilli())
if p.Dividends["TCB"][event.ProviderID].VNDPerShare != 1500 {
t.Fatal("failed future notice removed stored history")
}
}
func TestDividendCallbackUsesStoredEventAndCurrentHoldingOnce(t *testing.T) {
now := time.Date(2026, 6, 25, 12, 0, 0, 0, saigonLocation)
event := cashDividendEvent(now, startOfSaigonDay(now))
s, store, pending, rb, _, _ := newDividendFlowState(t, []DividendEvent{event})
seedDividendFlowPortfolio(t, store, now, 100)
for range 2 {
if err := s.handleStats(context.Background(), rb.Bot, testutil.NewPrivateMessage(7, "/stock_portfolio")); err != nil {
t.Fatal(err)
}
}
keys, err := pending.List(context.Background(), pendingDividendPrefix)
if err != nil || len(keys) != 2 {
t.Fatalf("pending keys = %v, err=%v", keys, err)
}
key := keys[0]
token := strings.TrimPrefix(key, pendingDividendPrefix)
action, _, _ := pending.Get(context.Background(), key)
if action.ProviderEventID != event.ProviderID || action.Symbol != "TCB" {
t.Fatalf("pending action = %+v", action)
}
if err := s.handleDividendCallback(context.Background(), rb.Bot, dividendCallbackUpdate(8, 7, action.MessageID, token)); err != nil {
t.Fatal(err)
}
p, _ := LoadPortfolio(context.Background(), store, 7, now.UnixMilli())
if p.VND != 100_000 {
t.Fatalf("expired action changed balance: %v", p.VND)
t.Fatalf("unauthorized click changed balance: %v", p.VND)
}
}
func TestDividendSuggestionFailureLeavesCursorUnchanged(t *testing.T) {
event := DividendEvent{
ProviderID: "2612974", Symbol: "TCB", Kind: DividendKindCash,
PublishedAt: time.Date(2026, 6, 24, 8, 0, 0, 0, saigonLocation), VNDPerShare: 1500,
}
s, store, _, rb, now := newDividendFlowState(t, []DividendEvent{event})
seedDividendFlowPortfolio(t, store, now, 100)
p, _ := LoadPortfolio(context.Background(), store, 7, now.UnixMilli())
before := p.Assets["TCB"].DividendCheckedAt
rb.FailMethod("sendMessage", http.StatusInternalServerError, "")
_ = s.notifyDividendEvents(context.Background(), rb.Bot, testutil.NewPrivateMessage(7, "/stock_portfolio").Message, 7, p, now)
p, _ = LoadPortfolio(context.Background(), store, 7, now.UnixMilli())
if got := p.Assets["TCB"].DividendCheckedAt; got != before {
t.Fatalf("failed delivery advanced cursor: got %d want %d", got, before)
}
}
func TestAdvanceDividendCursorPreservesConcurrentTrade(t *testing.T) {
s, store, _, _, now := newDividendFlowState(t, nil)
seedDividendFlowPortfolio(t, store, now, 100)
p, _ := LoadPortfolio(context.Background(), store, 7, now.UnixMilli())
if err := p.BuyTicker("TCB", 25, 1_000_000, now.UnixMilli()); err != nil {
if err := s.handleDividendCallback(context.Background(), rb.Bot, dividendCallbackUpdate(7, 7, action.MessageID, token)); err != nil {
t.Fatal(err)
}
p, _ = LoadPortfolio(context.Background(), store, 7, now.UnixMilli())
if p.VND != 250_000 || !p.Dividends["TCB"][event.ProviderID].Processed {
t.Fatalf("processed portfolio = %+v", p)
}
if err := s.handleDividendCallback(context.Background(), rb.Bot, dividendCallbackUpdate(7, 7, action.MessageID, token)); err != nil {
t.Fatal(err)
}
secondAction, _, _ := pending.Get(context.Background(), keys[1])
secondToken := strings.TrimPrefix(keys[1], pendingDividendPrefix)
if err := s.handleDividendCallback(context.Background(), rb.Bot, dividendCallbackUpdate(7, 7, secondAction.MessageID, secondToken)); err != nil {
t.Fatal(err)
}
p, _ = LoadPortfolio(context.Background(), store, 7, now.UnixMilli())
if p.VND != 250_000 {
t.Fatalf("second valid button credited twice: %v", p.VND)
}
}
func TestStoredDueEventStillNotifiesWhenSSIOmitsIt(t *testing.T) {
now := time.Date(2026, 6, 25, 12, 0, 0, 0, saigonLocation)
s, store, pending, rb, _, _ := newDividendFlowState(t, nil)
seedDividendFlowPortfolio(t, store, now, 100)
p, _ := LoadPortfolio(context.Background(), store, 7, now.UnixMilli())
p.Dividends["TCB"] = map[string]DividendRecord{
"2612974": {
Kind: DividendKindCash, PublishedAt: now.AddDate(0, 0, -45).UnixMilli(),
RecordDate: startOfSaigonDay(now).UnixMilli(), VNDPerShare: 1500,
},
}
if err := SavePortfolio(context.Background(), store, 7, p); err != nil {
t.Fatal(err)
}
if err := s.advanceDividendCursors(context.Background(), 7, []string{"TCB"}, now.UnixMilli()); err != nil {
if err := s.handleStats(context.Background(), rb.Bot, testutil.NewPrivateMessage(7, "/stock_portfolio")); err != nil {
t.Fatal(err)
}
p, _ = LoadPortfolio(context.Background(), store, 7, now.UnixMilli())
if p.Assets["TCB"].Quantity != 125 || p.Assets["TCB"].Base != 4_000_000 {
t.Fatalf("cursor merge lost trade: %+v", p.Assets["TCB"])
if calls := rb.Sent(); len(calls) != 2 || !strings.Contains(calls[1].Text(), "Dividend ready to apply") {
t.Fatalf("messages = %#v", calls)
}
keys, err := pending.List(context.Background(), pendingDividendPrefix)
if err != nil || len(keys) != 1 {
t.Fatalf("pending = %v, err=%v", keys, err)
}
}
func TestConcurrentDividendSuggestionsCreateOneButton(t *testing.T) {
s, store, pending, rb, now := newDividendFlowState(t, nil)
func TestStoredFutureEventStillNotifiesWhenSSIOmitsIt(t *testing.T) {
now := time.Date(2026, 6, 25, 12, 0, 0, 0, saigonLocation)
s, store, pending, rb, _, _ := newDividendFlowState(t, nil)
seedDividendFlowPortfolio(t, store, now, 100)
event := DividendEvent{ProviderID: "2612974", Symbol: "TCB", Kind: DividendKindCash, VNDPerShare: 1500}
p, _ := LoadPortfolio(context.Background(), store, 7, now.UnixMilli())
p.Dividends["TCB"] = map[string]DividendRecord{
"2612974": {
Kind: DividendKindCash, PublishedAt: now.Add(-24 * time.Hour).UnixMilli(),
RecordDate: now.AddDate(0, 0, 5).UnixMilli(), VNDPerShare: 1500,
},
}
if err := SavePortfolio(context.Background(), store, 7, p); err != nil {
t.Fatal(err)
}
if err := s.handleStats(context.Background(), rb.Bot, testutil.NewPrivateMessage(7, "/stock_portfolio")); err != nil {
t.Fatal(err)
}
if calls := rb.Sent(); len(calls) != 2 || !strings.Contains(calls[1].Text(), "Upcoming dividend event") {
t.Fatalf("messages = %#v", calls)
}
keys, err := pending.List(context.Background(), pendingDividendPrefix)
if err != nil || len(keys) != 0 {
t.Fatalf("pending = %v, err=%v", keys, err)
}
}
func TestDividendCallbackAppliesShareEventFromStoredHistory(t *testing.T) {
now := time.Date(2026, 6, 25, 12, 0, 0, 0, saigonLocation)
s, store, pending, rb, _, _ := newDividendFlowState(t, nil)
seedDividendFlowPortfolio(t, store, now, 139)
p, _ := LoadPortfolio(context.Background(), store, 7, now.UnixMilli())
p.Dividends["TCB"] = map[string]DividendRecord{
"2612975": {
Kind: DividendKindShares, PublishedAt: now.Add(-24 * time.Hour).UnixMilli(),
RecordDate: startOfSaigonDay(now).UnixMilli(), OwnedShares: 100, NewShares: 10,
},
}
if err := SavePortfolio(context.Background(), store, 7, p); err != nil {
t.Fatal(err)
}
action := PendingDividendAction{
OwnerUserID: 7, ChatID: 7, MessageID: 1, ProviderEventID: "2612975", Symbol: "TCB",
PositionOpenedAt: p.Assets["TCB"].OpenedAt, CreatedAt: now.UnixMilli(), ExpiresAt: now.Add(time.Hour).UnixMilli(),
}
if err := pending.Put(context.Background(), pendingDividendKey("abcdefghijklmnopqrstuv"), action); err != nil {
t.Fatal(err)
}
if err := s.handleDividendCallback(context.Background(), rb.Bot, dividendCallbackUpdate(7, 7, 1, "abcdefghijklmnopqrstuv")); err != nil {
t.Fatal(err)
}
p, _ = LoadPortfolio(context.Background(), store, 7, now.UnixMilli())
if p.Assets["TCB"].Quantity != 152 || p.Assets["TCB"].Base != 139*30_000 || !p.Dividends["TCB"]["2612975"].Processed {
t.Fatalf("share dividend result = %+v", p)
}
}
func TestExpiredDividendActionDoesNotApply(t *testing.T) {
now := time.Date(2026, 6, 25, 12, 0, 0, 0, saigonLocation)
s, store, pending, rb, _, _ := newDividendFlowState(t, nil)
seedDividendFlowPortfolio(t, store, now, 100)
p, _ := LoadPortfolio(context.Background(), store, 7, now.UnixMilli())
p.Dividends["TCB"] = map[string]DividendRecord{
"2612974": {Kind: DividendKindCash, PublishedAt: now.Add(-24 * time.Hour).UnixMilli(), RecordDate: startOfSaigonDay(now).UnixMilli(), VNDPerShare: 1500},
}
if err := SavePortfolio(context.Background(), store, 7, p); err != nil {
t.Fatal(err)
}
action := PendingDividendAction{
OwnerUserID: 7, ChatID: 7, MessageID: 1, ProviderEventID: "2612974", Symbol: "TCB",
PositionOpenedAt: p.Assets["TCB"].OpenedAt, CreatedAt: now.Add(-2 * time.Hour).UnixMilli(), ExpiresAt: now.Add(-time.Hour).UnixMilli(),
}
key := pendingDividendKey("abcdefghijklmnopqrstuv")
if err := pending.Put(context.Background(), key, action); err != nil {
t.Fatal(err)
}
if err := s.handleDividendCallback(context.Background(), rb.Bot, dividendCallbackUpdate(7, 7, 1, "abcdefghijklmnopqrstuv")); err != nil {
t.Fatal(err)
}
p, _ = LoadPortfolio(context.Background(), store, 7, now.UnixMilli())
if p.VND != 100_000 || p.Dividends["TCB"]["2612974"].Processed {
t.Fatalf("expired action changed portfolio: %+v", p)
}
if _, _, err := pending.Get(context.Background(), key); !errors.Is(err, storage.ErrNotFound) {
t.Fatalf("expired action was not deleted: %v", err)
}
}
func TestConcurrentPortfolioRequestsCreateSeparatePendingActions(t *testing.T) {
now := time.Date(2026, 6, 25, 12, 0, 0, 0, saigonLocation)
s, store, pending, rb, _, _ := newDividendFlowState(t, nil)
seedDividendFlowPortfolio(t, store, now, 100)
p, _ := LoadPortfolio(context.Background(), store, 7, now.UnixMilli())
p.Dividends["TCB"] = map[string]DividendRecord{
"2612974": {Kind: DividendKindCash, PublishedAt: now.Add(-24 * time.Hour).UnixMilli(), RecordDate: startOfSaigonDay(now).UnixMilli(), VNDPerShare: 1500},
}
if err := SavePortfolio(context.Background(), store, 7, p); err != nil {
t.Fatal(err)
}
msg := testutil.NewPrivateMessage(7, "/stock_portfolio").Message
ref := dividendRef{symbol: "TCB", eventID: "2612974"}
var wg sync.WaitGroup
errs := make(chan error, 2)
for range 2 {
wg.Add(1)
go func() {
defer wg.Done()
errs <- s.sendDividendSuggestion(context.Background(), rb.Bot, msg, 7, now.Add(-48*time.Hour).UnixMilli(), now, event)
errs <- s.sendDividendSuggestion(context.Background(), rb.Bot, msg, 7, p.Assets["TCB"].OpenedAt, ref)
}()
}
wg.Wait()
@@ -365,54 +450,60 @@ func TestConcurrentDividendSuggestionsCreateOneButton(t *testing.T) {
t.Fatal(err)
}
}
if calls := rb.Sent(); len(calls) != 1 {
t.Fatalf("sent %d duplicate suggestions: %#v", len(calls), calls)
}
keys, err := pending.List(context.Background(), pendingDividendPrefix)
if err != nil || len(keys) != 1 {
t.Fatalf("pending actions = %v, err=%v", keys, err)
if err != nil || len(keys) != 2 || len(rb.Sent()) != 2 {
t.Fatalf("pending=%v messages=%d err=%v", keys, len(rb.Sent()), err)
}
}
func TestDividendFetchDoesNotBindOldEventToReopenedPosition(t *testing.T) {
s, store, pending, rb, now := newDividendFlowState(t, nil)
func TestDividendCallbackRejectsPositionOpenedAfterRecordDate(t *testing.T) {
now := time.Date(2026, 6, 25, 12, 0, 0, 0, saigonLocation)
s, store, pending, rb, _, _ := newDividendFlowState(t, nil)
seedDividendFlowPortfolio(t, store, now, 100)
blocking := &blockingDividendProvider{
started: make(chan struct{}), release: make(chan struct{}),
events: []DividendEvent{{ProviderID: "2612974", Symbol: "TCB", Kind: DividendKindCash, VNDPerShare: 1500}},
}
s.dividends = blocking
done := make(chan error, 1)
go func() {
done <- s.handleStats(context.Background(), rb.Bot, testutil.NewPrivateMessage(7, "/stock_portfolio"))
}()
<-blocking.started
p, _ := LoadPortfolio(context.Background(), store, 7, now.UnixMilli())
if _, _, ok, err := p.SellTicker("TCB", 100); err != nil || !ok {
t.Fatalf("sell during fetch: ok=%v err=%v", ok, err)
}
if err := p.BuyTicker("TCB", 50, 2_000_000, now.Add(time.Minute).UnixMilli()); err != nil {
t.Fatal(err)
recordDate := startOfSaigonDay(now.AddDate(0, 0, -1))
position := p.Assets["TCB"]
position.OpenedAt = now.UnixMilli()
p.Assets["TCB"] = position
p.Dividends["TCB"] = map[string]DividendRecord{
"2612974": {Kind: DividendKindCash, PublishedAt: now.AddDate(0, 0, -2).UnixMilli(), RecordDate: recordDate.UnixMilli(), VNDPerShare: 1500},
}
if err := SavePortfolio(context.Background(), store, 7, p); err != nil {
t.Fatal(err)
}
close(blocking.release)
if err := <-done; err != nil {
action := PendingDividendAction{
OwnerUserID: 7, ChatID: 7, MessageID: 1, ProviderEventID: "2612974", Symbol: "TCB",
PositionOpenedAt: position.OpenedAt, CreatedAt: now.UnixMilli(), ExpiresAt: now.Add(time.Hour).UnixMilli(),
}
key := pendingDividendKey("abcdefghijklmnopqrstuv")
if err := pending.Put(context.Background(), key, action); err != nil {
t.Fatal(err)
}
if calls := rb.Sent(); len(calls) != 1 || !strings.Contains(calls[0].Text(), "Stock Portfolio") {
t.Fatalf("old lifecycle event was sent: %#v", calls)
if err := s.handleDividendCallback(context.Background(), rb.Bot, dividendCallbackUpdate(7, 7, 1, "abcdefghijklmnopqrstuv")); err != nil {
t.Fatal(err)
}
keys, err := pending.List(context.Background(), pendingDividendPrefix)
if err != nil || len(keys) != 0 {
t.Fatalf("old lifecycle created pending action: %v, err=%v", keys, err)
p, _ = LoadPortfolio(context.Background(), store, 7, now.UnixMilli())
if p.VND != 100_000 || p.Dividends["TCB"]["2612974"].Processed {
t.Fatalf("ineligible position applied event: %+v", p)
}
}
func dividendCallbackUpdate(userID, chatID int64, messageID int) *models.Update {
func TestProviderFailureKeepsPortfolioAvailable(t *testing.T) {
s, store, _, rb, now, provider := newDividendFlowState(t, nil)
provider.err = errors.New("SSI unavailable")
seedDividendFlowPortfolio(t, store, now, 100)
if err := s.handleStats(context.Background(), rb.Bot, testutil.NewPrivateMessage(7, "/stock_portfolio")); err != nil {
t.Fatal(err)
}
calls := rb.Sent()
if len(calls) != 2 || !strings.Contains(calls[0].Text(), "Stock Portfolio") || !strings.Contains(calls[1].Text(), "could not be checked") {
t.Fatalf("messages = %#v", calls)
}
}
func dividendCallbackUpdate(userID, chatID int64, messageID int, token string) *models.Update {
return &models.Update{CallbackQuery: &models.CallbackQuery{
ID: "query", From: models.User{ID: userID}, Data: dividendCallbackPrefix + "abcdefghijklmnopqrstuv",
ID: "query", From: models.User{ID: userID}, Data: dividendCallbackPrefix + token,
Message: models.MaybeInaccessibleMessage{Type: models.MaybeInaccessibleMessageTypeMessage, Message: &models.Message{
ID: messageID, Chat: models.Chat{ID: chatID, Type: models.ChatTypePrivate},
}},
internal/modules/stock/dividend_history.go
+146
View File
@@ -0,0 +1,146 @@
package stock
import "time"
const (
dividendDiscoveryWindow = 30 * 24 * time.Hour
dividendRetentionDays = 90
)
func (r DividendRecord) valid() bool {
if r.PublishedAt <= 0 || r.ExDate < 0 || r.RecordDate < 0 || r.PaymentDate < 0 {
return false
}
switch r.Kind {
case DividendKindCash:
return r.VNDPerShare > 0 && r.OwnedShares == 0 && r.NewShares == 0
case DividendKindShares:
return r.VNDPerShare == 0 && r.OwnedShares > 0 && r.NewShares > 0
default:
return false
}
}
func dividendRecordFromEvent(event DividendEvent) DividendRecord {
return DividendRecord{
Kind: event.Kind,
PublishedAt: timeMillis(event.PublishedAt),
ExDate: timeMillis(event.ExDate),
RecordDate: timeMillis(event.RecordDate),
PaymentDate: timeMillis(event.PaymentDate),
VNDPerShare: event.VNDPerShare,
OwnedShares: event.OwnedShares,
NewShares: event.NewShares,
Title: event.Title,
SourceURL: event.SourceURL,
}
}
func (r DividendRecord) event(symbol, providerID string) DividendEvent {
return DividendEvent{
ProviderID: providerID,
Symbol: symbol,
Kind: r.Kind,
PublishedAt: millisTime(r.PublishedAt),
ExDate: millisTime(r.ExDate),
RecordDate: millisTime(r.RecordDate),
PaymentDate: millisTime(r.PaymentDate),
VNDPerShare: r.VNDPerShare,
OwnedShares: r.OwnedShares,
NewShares: r.NewShares,
Title: r.Title,
SourceURL: r.SourceURL,
}
}
func timeMillis(value time.Time) int64 {
if value.IsZero() {
return 0
}
return value.UnixMilli()
}
func millisTime(value int64) time.Time {
if value <= 0 {
return time.Time{}
}
return time.UnixMilli(value).In(saigonLocation)
}
func (r DividendRecord) retentionAnchor() int64 {
if r.RecordDate != 0 {
return r.RecordDate
}
return r.PublishedAt
}
func (p Portfolio) dividendRecord(symbol, eventID string) (DividendRecord, bool) {
events := p.Dividends[symbol]
if events == nil {
return DividendRecord{}, false
}
record, exists := events[eventID]
return record, exists
}
func (p *Portfolio) setDividendRecord(symbol, eventID string, record DividendRecord) {
if p.Dividends == nil {
p.Dividends = map[string]map[string]DividendRecord{}
}
events := p.Dividends[symbol]
if events == nil {
events = map[string]DividendRecord{}
p.Dividends[symbol] = events
}
events[eventID] = record
}
func (p *Portfolio) upsertDividendEvent(event DividendEvent) bool {
old, exists := p.dividendRecord(event.Symbol, event.ProviderID)
updated := dividendRecordFromEvent(event)
updated.Processed = old.Processed
if exists && old == updated {
return false
}
p.setDividendRecord(event.Symbol, event.ProviderID, updated)
return true
}
func (p *Portfolio) pruneDividendHistory(now time.Time) bool {
changed := false
for symbol, events := range p.Dividends {
for eventID, event := range events {
if dividendRecordExpired(event, now) {
delete(events, eventID)
changed = true
}
}
if len(events) == 0 {
delete(p.Dividends, symbol)
}
}
return changed
}
func dividendRecordExpired(event DividendRecord, now time.Time) bool {
anchor := event.retentionAnchor()
return anchor > 0 && !startOfSaigonDay(now).Before(startOfSaigonDay(millisTime(anchor)).AddDate(0, 0, dividendRetentionDays))
}
func dividendRecordDue(event DividendRecord, now time.Time) bool {
return event.RecordDate > 0 && !startOfSaigonDay(now).Before(startOfSaigonDay(millisTime(event.RecordDate)))
}
func positionOpenedByRecordDate(position AssetPosition, event DividendRecord) bool {
if event.RecordDate <= 0 || position.OpenedAt < 0 {
return false
}
// OpenedAt was added after portfolios already existed. Zero means the
// lifecycle predates tracking, so preserve eligibility for legacy holdings.
if position.OpenedAt == 0 {
return true
}
openedDay := startOfSaigonDay(millisTime(position.OpenedAt))
recordDay := startOfSaigonDay(millisTime(event.RecordDate))
return !openedDay.After(recordDay)
}
internal/modules/stock/dividend_history_test.go
+65
View File
@@ -0,0 +1,65 @@
package stock
import (
"testing"
"time"
)
func TestUpsertDividendEventPreservesLocalStateAndProviderCorrections(t *testing.T) {
p := NewPortfolio(1)
p.Dividends["TCB"] = map[string]DividendRecord{
"2612974": {
Kind: DividendKindCash, PublishedAt: 1, VNDPerShare: 1000,
Processed: true,
},
}
event := DividendEvent{
ProviderID: "2612974", Symbol: "TCB", Kind: DividendKindCash,
PublishedAt: time.UnixMilli(2), RecordDate: time.UnixMilli(3), VNDPerShare: 1500,
}
if !p.upsertDividendEvent(event) {
t.Fatal("provider correction was not recorded")
}
got := p.Dividends["TCB"]["2612974"]
if got.VNDPerShare != 1500 || got.RecordDate != 3 || !got.Processed {
t.Fatalf("merged event = %+v", got)
}
}
func TestPruneDividendHistoryUsesRecordDateThenPublishedDate(t *testing.T) {
now := time.Date(2026, 7, 22, 12, 0, 0, 0, saigonLocation)
p := NewPortfolio(1)
p.Dividends["TCB"] = map[string]DividendRecord{
"record-expired": {
Kind: DividendKindCash, PublishedAt: now.AddDate(0, 0, -120).UnixMilli(),
RecordDate: now.AddDate(0, 0, -90).UnixMilli(), VNDPerShare: 1000,
},
"missing-expired": {
Kind: DividendKindCash, PublishedAt: now.AddDate(0, 0, -90).UnixMilli(), VNDPerShare: 1000,
},
"keep": {
Kind: DividendKindCash, PublishedAt: now.AddDate(0, 0, -100).UnixMilli(),
RecordDate: now.AddDate(0, 0, -89).UnixMilli(), VNDPerShare: 1000,
},
}
if !p.pruneDividendHistory(now) {
t.Fatal("prune reported no change")
}
if len(p.Dividends["TCB"]) != 1 || p.Dividends["TCB"]["keep"].RecordDate == 0 {
t.Fatalf("retained history = %+v", p.Dividends)
}
}
func TestPositionOpenedByRecordDateUsesSaigonCalendarDay(t *testing.T) {
recordDate := time.Date(2026, 6, 25, 0, 0, 0, 0, saigonLocation)
event := DividendRecord{RecordDate: recordDate.UnixMilli()}
if !positionOpenedByRecordDate(AssetPosition{}, event) {
t.Fatal("legacy position without openedAt should remain eligible")
}
if !positionOpenedByRecordDate(AssetPosition{OpenedAt: recordDate.Add(12 * time.Hour).UnixMilli()}, event) {
t.Fatal("position opened on record date should be eligible")
}
if positionOpenedByRecordDate(AssetPosition{OpenedAt: recordDate.AddDate(0, 0, 1).UnixMilli()}, event) {
t.Fatal("position opened after record date should be ineligible")
}
}
internal/modules/stock/dividend_notifications.go
+240 -164
View File
@@ -23,118 +23,77 @@ const (
dividendFetchWorkers = 4
)
type dividendCheckResult struct {
symbol string
openedAt int64
after time.Time
events []DividendEvent
err error
type dividendRef struct {
symbol string
eventID string
}
func (s *state) notifyDividendEvents(ctx context.Context, b *bot.Bot, msg *models.Message, userID int64, p Portfolio, checkedThrough time.Time) error {
if s.dividends == nil || s.pending == nil || len(p.Assets) == 0 {
return nil
}
s.cleanupExpiredDividends(ctx, checkedThrough.UnixMilli())
type dividendFetchJob struct {
symbol string
after time.Time
through time.Time
recent bool
targetIDs map[string]struct{}
}
type holding struct {
symbol string
qty int64
after time.Time
type dividendCheckResult struct {
job dividendFetchJob
events []DividendEvent
err error
}
func (job dividendFetchJob) includes(event DividendEvent) bool {
if event.Symbol != job.symbol || event.PublishedAt.Before(job.after) || event.PublishedAt.After(job.through) {
return false
}
holdings := make([]holding, 0, len(p.Assets))
for symbol, position := range p.Assets {
if position.Quantity > 0 {
holdings = append(holdings, holding{
symbol: symbol,
qty: position.Quantity,
after: time.UnixMilli(position.DividendCheckedAt),
})
if job.recent {
return true
}
_, wanted := job.targetIDs[event.ProviderID]
return wanted
}
func (s *state) notifyDividendEvents(ctx context.Context, b *bot.Bot, msg *models.Message, userID int64, snapshot Portfolio, checkedThrough time.Time) error {
if s.pending != nil {
s.cleanupExpiredDividends(ctx, checkedThrough.UnixMilli())
}
failed, err := s.syncDividendHistory(ctx, userID, snapshot, checkedThrough)
if err != nil {
log.Error("stock_save_dividend_history", "user", userID, "err", err)
if replyErr := chathelper.Reply(ctx, b, msg, "Dividend events were checked, but the history could not be saved. Try again later."); replyErr != nil {
return replyErr
}
}
if len(holdings) == 0 {
return nil
}
sort.Slice(holdings, func(i, j int) bool { return holdings[i].symbol < holdings[j].symbol })
fetchCtx, cancel := context.WithTimeout(ctx, dividendFetchTimeout)
defer cancel()
results := make([]dividendCheckResult, len(holdings))
sem := make(chan struct{}, dividendFetchWorkers)
var wg sync.WaitGroup
for index, h := range holdings {
index, h := index, h
wg.Add(1)
go func() {
defer wg.Done()
sem <- struct{}{}
defer func() { <-sem }()
events, err := s.dividends.FetchDividendEvents(fetchCtx, h.symbol, h.after, checkedThrough)
results[index] = dividendCheckResult{
symbol: h.symbol,
openedAt: p.Assets[h.symbol].OpenedAt,
after: h.after,
events: events,
err: err,
}
}()
latest, err := LoadPortfolio(ctx, s.store, userID, checkedThrough.UnixMilli())
if err != nil {
return err
}
wg.Wait()
successful := make([]string, 0, len(results))
failed := make([]string, 0)
for _, result := range results {
if result.err != nil {
log.Error("stock_dividend_events_checked",
"user", userID,
"ticker", result.symbol,
"from", result.after.UnixMilli(),
"through", checkedThrough.UnixMilli(),
"events", 0,
"status", "error",
"err", result.err)
failed = append(failed, result.symbol)
refs := sortedDividendRefs(latest)
for _, ref := range refs {
position, held := latest.Assets[ref.symbol]
if !held || position.Quantity <= 0 {
continue
}
log.Info("stock_dividend_events_checked",
"user", userID,
"ticker", result.symbol,
"from", result.after.UnixMilli(),
"through", checkedThrough.UnixMilli(),
"events", len(result.events),
"status", "success")
sort.Slice(result.events, func(i, j int) bool {
if result.events[i].PublishedAt.Equal(result.events[j].PublishedAt) {
return result.events[i].ProviderID < result.events[j].ProviderID
}
return result.events[i].PublishedAt.Before(result.events[j].PublishedAt)
})
delivered := true
for _, event := range result.events {
if _, applied := p.AppliedDividendEvents[dividendLedgerKey(event.ProviderID)]; applied {
continue
}
if err := s.sendDividendSuggestion(ctx, b, msg, userID, result.openedAt, checkedThrough, event); err != nil {
log.Error("stock_send_dividend_suggestion", "user", userID, "ticker", result.symbol, "event", event.ProviderID, "err", err)
delivered = false
break
}
record, exists := latest.dividendRecord(ref.symbol, ref.eventID)
if !exists || record.Processed {
continue
}
if delivered {
successful = append(successful, result.symbol)
} else {
failed = append(failed, result.symbol)
if !dividendRecordDue(record, checkedThrough) {
if err := s.sendFutureDividendNotice(ctx, b, msg, userID, position.OpenedAt, ref); err != nil {
log.Error("stock_send_dividend_notice", "user", userID, "ticker", ref.symbol, "event", ref.eventID, "err", err)
failed = append(failed, ref.symbol)
}
continue
}
if err := s.sendDividendSuggestion(ctx, b, msg, userID, position.OpenedAt, ref); err != nil {
log.Error("stock_send_dividend_suggestion", "user", userID, "ticker", ref.symbol, "event", ref.eventID, "err", err)
failed = append(failed, ref.symbol)
}
}
if len(successful) > 0 {
if err := s.advanceDividendCursors(ctx, userID, successful, checkedThrough.UnixMilli()); err != nil {
log.Error("stock_save_dividend_cursors", "user", userID, "err", err)
if replyErr := chathelper.Reply(ctx, b, msg, "Dividend events were checked, but the check time could not be saved. You may see the same suggestions again."); replyErr != nil {
return replyErr
}
}
}
if len(failed) > 0 {
sort.Strings(failed)
failed = uniqueStrings(failed)
@@ -143,64 +102,196 @@ func (s *state) notifyDividendEvents(ctx context.Context, b *bot.Bot, msg *model
return nil
}
func (s *state) advanceDividendCursors(ctx context.Context, userID int64, symbols []string, checkedThrough int64) error {
func (s *state) syncDividendHistory(ctx context.Context, userID int64, snapshot Portfolio, now time.Time) ([]string, error) {
jobs := dividendFetchJobs(snapshot, now)
results := s.fetchDividendJobs(ctx, jobs)
failed := make([]string, 0)
defer s.locks.Acquire(strconv.FormatInt(userID, 10))()
p, err := LoadPortfolio(ctx, s.store, userID, checkedThrough)
latest, err := LoadPortfolio(ctx, s.store, userID, now.UnixMilli())
if err != nil {
return err
return nil, err
}
changed := false
for _, symbol := range symbols {
position, ok := p.Assets[symbol]
if ok && position.DividendCheckedAt < checkedThrough {
position.DividendCheckedAt = checkedThrough
p.Assets[symbol] = position
changed = true
changed := latest.pruneDividendHistory(now)
for _, result := range results {
if result.err != nil {
log.Error("stock_dividend_events_checked",
"user", userID,
"ticker", result.job.symbol,
"from", result.job.after.UnixMilli(),
"through", result.job.through.UnixMilli(),
"events", 0,
"status", "error",
"err", result.err)
failed = append(failed, result.job.symbol)
continue
}
log.Info("stock_dividend_events_checked",
"user", userID,
"ticker", result.job.symbol,
"from", result.job.after.UnixMilli(),
"through", result.job.through.UnixMilli(),
"events", len(result.events),
"status", "success")
for _, event := range result.events {
if !result.job.includes(event) {
continue
}
if dividendRecordExpired(dividendRecordFromEvent(event), now) {
continue
}
if latest.upsertDividendEvent(event) {
changed = true
}
}
}
if !changed {
return nil
if changed {
if err := SavePortfolio(ctx, s.store, userID, latest); err != nil {
return failed, err
}
}
return SavePortfolio(ctx, s.store, userID, p)
return failed, nil
}
func (s *state) sendDividendSuggestion(ctx context.Context, b *bot.Bot, msg *models.Message, userID, expectedOpenedAt int64, checkedThrough time.Time, event DividendEvent) error {
func dividendFetchJobs(p Portfolio, now time.Time) []dividendFetchJob {
recentAfter := now.Add(-dividendDiscoveryWindow)
jobs := make([]dividendFetchJob, 0, len(p.Assets))
for symbol, position := range p.Assets {
if position.Quantity > 0 {
jobs = append(jobs, dividendFetchJob{symbol: symbol, after: recentAfter, through: now, recent: true})
}
}
type historicalKey struct {
symbol string
day int64
}
historical := map[historicalKey]map[string]struct{}{}
for symbol, events := range p.Dividends {
if position, held := p.Assets[symbol]; !held || position.Quantity <= 0 {
continue
}
for eventID, event := range events {
if event.Processed || event.PublishedAt <= 0 || !millisTime(event.PublishedAt).Before(recentAfter) {
continue
}
if event.RecordDate != 0 && !dividendRecordDue(event, now) {
continue
}
day := startOfSaigonDay(millisTime(event.PublishedAt))
key := historicalKey{symbol: symbol, day: day.UnixMilli()}
if historical[key] == nil {
historical[key] = map[string]struct{}{}
}
historical[key][eventID] = struct{}{}
}
}
for key, targetIDs := range historical {
day := millisTime(key.day)
jobs = append(jobs, dividendFetchJob{
symbol: key.symbol, after: day, through: day.Add(24*time.Hour - time.Millisecond), targetIDs: targetIDs,
})
}
sort.Slice(jobs, func(i, j int) bool {
if jobs[i].symbol != jobs[j].symbol {
return jobs[i].symbol < jobs[j].symbol
}
return jobs[i].after.Before(jobs[j].after)
})
return jobs
}
func (s *state) fetchDividendJobs(ctx context.Context, jobs []dividendFetchJob) []dividendCheckResult {
if len(jobs) == 0 || s.dividends == nil {
return nil
}
results := make([]dividendCheckResult, len(jobs))
fetchCtx, cancel := context.WithTimeout(ctx, dividendFetchTimeout)
defer cancel()
sem := make(chan struct{}, dividendFetchWorkers)
var wg sync.WaitGroup
for index, job := range jobs {
index, job := index, job
wg.Add(1)
go func() {
defer wg.Done()
sem <- struct{}{}
defer func() { <-sem }()
events, err := s.dividends.FetchDividendEvents(fetchCtx, job.symbol, job.after, job.through)
results[index] = dividendCheckResult{job: job, events: events, err: err}
}()
}
wg.Wait()
return results
}
func sortedDividendRefs(p Portfolio) []dividendRef {
refs := make([]dividendRef, 0)
for symbol, events := range p.Dividends {
for eventID := range events {
refs = append(refs, dividendRef{symbol: symbol, eventID: eventID})
}
}
sort.Slice(refs, func(i, j int) bool {
left := p.Dividends[refs[i].symbol][refs[i].eventID]
right := p.Dividends[refs[j].symbol][refs[j].eventID]
if left.PublishedAt != right.PublishedAt {
return left.PublishedAt < right.PublishedAt
}
if refs[i].symbol != refs[j].symbol {
return refs[i].symbol < refs[j].symbol
}
return refs[i].eventID < refs[j].eventID
})
return refs
}
func (s *state) sendFutureDividendNotice(ctx context.Context, b *bot.Bot, msg *models.Message, userID, expectedOpenedAt int64, ref dividendRef) error {
defer s.locks.Acquire(strconv.FormatInt(userID, 10))()
now := s.now()
latest, err := LoadPortfolio(ctx, s.store, userID, now.UnixMilli())
p, err := LoadPortfolio(ctx, s.store, userID, now.UnixMilli())
if err != nil {
return err
}
if _, applied := latest.AppliedDividendEvents[dividendLedgerKey(event.ProviderID)]; applied {
record, exists := p.dividendRecord(ref.symbol, ref.eventID)
if !exists || record.Processed || dividendRecordDue(record, now) {
return nil
}
position, held := latest.Assets[event.Symbol]
position, held := p.Assets[ref.symbol]
if !held || position.Quantity <= 0 || position.OpenedAt != expectedOpenedAt {
return nil
}
observedHolding := position.Quantity
hasPending, err := s.hasPendingDividendEvent(ctx, userID, event.ProviderID, now.UnixMilli())
if _, err := b.SendMessage(ctx, &bot.SendMessageParams{
ChatID: msg.Chat.ID,
MessageThreadID: msg.MessageThreadID,
Text: dividendEventText(record.event(ref.symbol, ref.eventID), position.Quantity, false),
ParseMode: models.ParseModeHTML,
}); err != nil {
return err
}
return nil
}
func (s *state) sendDividendSuggestion(ctx context.Context, b *bot.Bot, msg *models.Message, userID, expectedOpenedAt int64, ref dividendRef) error {
defer s.locks.Acquire(strconv.FormatInt(userID, 10))()
now := s.now()
p, err := LoadPortfolio(ctx, s.store, userID, now.UnixMilli())
if err != nil {
return err
}
if hasPending {
record, exists := p.dividendRecord(ref.symbol, ref.eventID)
if !exists || record.Processed || !dividendRecordDue(record, now) {
return nil
}
position, held := p.Assets[ref.symbol]
if !held || position.Quantity <= 0 || position.OpenedAt != expectedOpenedAt || !positionOpenedByRecordDate(position, record) {
return nil
}
action := PendingDividendAction{
OwnerUserID: userID,
ChatID: msg.Chat.ID,
ProviderEventID: event.ProviderID,
Symbol: event.Symbol,
Kind: event.Kind,
VNDPerShare: event.VNDPerShare,
OwnedShares: event.OwnedShares,
NewShares: event.NewShares,
ObservedHolding: observedHolding,
PositionOpenedAt: position.OpenedAt,
CheckThrough: checkedThrough.UnixMilli(),
CreatedAt: now.UnixMilli(),
ExpiresAt: now.Add(pendingDividendTTL).UnixMilli(),
OwnerUserID: userID, ChatID: msg.Chat.ID, ProviderEventID: ref.eventID, Symbol: ref.symbol,
PositionOpenedAt: position.OpenedAt, CreatedAt: now.UnixMilli(), ExpiresAt: now.Add(pendingDividendTTL).UnixMilli(),
}
token, err := s.createPendingDividend(ctx, action)
if err != nil {
@@ -210,11 +301,10 @@ func (s *state) sendDividendSuggestion(ctx context.Context, b *bot.Bot, msg *mod
sent, err := b.SendMessage(ctx, &bot.SendMessageParams{
ChatID: msg.Chat.ID,
MessageThreadID: msg.MessageThreadID,
Text: dividendEventText(event, observedHolding),
Text: dividendEventText(record.event(ref.symbol, ref.eventID), position.Quantity, true),
ParseMode: models.ParseModeHTML,
ReplyMarkup: &models.InlineKeyboardMarkup{InlineKeyboard: [][]models.InlineKeyboardButton{{{
Text: "Apply dividend",
CallbackData: dividendCallbackPrefix + token,
Text: "Apply dividend", CallbackData: dividendCallbackPrefix + token,
}}}},
})
if err != nil {
@@ -230,28 +320,7 @@ func (s *state) sendDividendSuggestion(ctx context.Context, b *bot.Bot, msg *mod
return nil
}
func (s *state) hasPendingDividendEvent(ctx context.Context, userID int64, providerEventID string, now int64) (bool, error) {
keys, err := s.pending.List(ctx, pendingDividendPrefix)
if err != nil {
return false, err
}
for _, key := range keys {
action, _, getErr := s.pending.Get(ctx, key)
if getErr != nil {
continue
}
if action.ExpiresAt <= now || action.MessageID == 0 {
_ = s.pending.Delete(ctx, key)
continue
}
if action.OwnerUserID == userID && action.ProviderEventID == providerEventID {
return true, nil
}
}
return false, nil
}
func dividendEventText(event DividendEvent, observedHolding int64) string {
func dividendEventText(event DividendEvent, observedHolding int64, actionable bool) string {
var value string
switch event.Kind {
case DividendKindCash:
@@ -259,15 +328,18 @@ func dividendEventText(event DividendEvent, observedHolding int64) string {
case DividendKindShares:
value = strconv.FormatInt(event.OwnedShares, 10) + ":" + strconv.FormatInt(event.NewShares, 10) + " shares"
}
lines := []string{
"<b>Recent dividend event · " + html.EscapeString(event.Symbol) + "</b>",
html.EscapeString(value),
headline := "Upcoming dividend event · "
if actionable {
headline = "Dividend ready to apply · "
}
lines := []string{"<b>" + headline + html.EscapeString(event.Symbol) + "</b>", html.EscapeString(value)}
if !event.ExDate.IsZero() {
lines = append(lines, "Ex-right: "+event.ExDate.Format("02/01/2006"))
}
if !event.RecordDate.IsZero() {
lines = append(lines, "Record: "+event.RecordDate.Format("02/01/2006"))
} else {
lines = append(lines, "Record: awaiting SSI update")
}
if !event.PaymentDate.IsZero() {
lines = append(lines, "Payment/trading: "+event.PaymentDate.Format("02/01/2006"))
@@ -275,12 +347,16 @@ func dividendEventText(event DividendEvent, observedHolding int64) string {
if title := truncateRunes(strings.TrimSpace(event.Title), 240); title != "" {
lines = append(lines, html.EscapeString(title))
}
lines = append(lines,
"SSI event: <code>"+html.EscapeString(event.ProviderID)+"</code>",
"Observed holding: "+html.EscapeString(formatShareQuantity(observedHolding))+" shares.",
"The dividend uses your current holding when you accept.",
"Button expires in 24 hours.",
)
lines = append(lines, "SSI event: <code>"+html.EscapeString(event.ProviderID)+"</code>")
if actionable {
lines = append(lines,
"Current holding: "+html.EscapeString(formatShareQuantity(observedHolding))+" shares.",
"The dividend uses your current holding when you accept.",
"Button expires in 24 hours.",
)
} else {
lines = append(lines, "Approval becomes available from Record date.")
}
return strings.Join(lines, "\n")
}
internal/modules/stock/handlers.go
+6 -6
View File
@@ -311,7 +311,7 @@ func (s *state) handleCashDividend(ctx context.Context, b *bot.Bot, update *mode
return chathelper.Reply(ctx, b, update.Message, "Dividend amount is too large.")
}
if err := p.ApplyDividend(symbol, held, balance, s.now().UnixMilli()); err != nil {
return chathelper.Reply(ctx, b, update.Message, "Could not update dividend checkpoint. Try again later.")
return chathelper.Reply(ctx, b, update.Message, "Could not apply this dividend. Try again later.")
}
if err := SavePortfolio(ctx, s.store, userID, p); err != nil {
log.Error("stock_save_portfolio", "user", userID, "err", err)
@@ -374,7 +374,7 @@ func (s *state) handleShareDividend(ctx context.Context, b *bot.Bot, update *mod
return chathelper.Reply(ctx, b, update.Message, "Share dividend is too large.")
}
if err := p.ApplyDividend(symbol, finalHolding, p.VND, s.now().UnixMilli()); err != nil {
return chathelper.Reply(ctx, b, update.Message, "Could not update dividend checkpoint. Try again later.")
return chathelper.Reply(ctx, b, update.Message, "Could not apply this dividend. Try again later.")
}
if err := SavePortfolio(ctx, s.store, userID, p); err != nil {
log.Error("stock_save_portfolio", "user", userID, "err", err)
@@ -443,7 +443,7 @@ func (s *state) handleDividend(ctx context.Context, b *bot.Bot, update *models.U
}
if err := p.ApplyDividend(symbol, finalHolding, balance, s.now().UnixMilli()); err != nil {
return chathelper.Reply(ctx, b, update.Message, "Could not update dividend checkpoint. Try again later.")
return chathelper.Reply(ctx, b, update.Message, "Could not apply this dividend. Try again later.")
}
if err := SavePortfolio(ctx, s.store, userID, p); err != nil {
log.Error("stock_save_portfolio", "user", userID, "err", err)
@@ -457,9 +457,9 @@ func (s *state) handleDividend(ctx context.Context, b *bot.Bot, update *models.U
"\nBalance: "+FormatVND(balance))
}
// handleStats renders the portfolio first, then checks each held ticker for
// dividend events. Cursor persistence reloads and merges under the user lock,
// so network calls never block concurrent portfolio mutations.
// handleStats renders the portfolio first, then synchronizes dividend history
// for each held ticker. Network calls never hold the user lock, while history
// merging and notification state updates reload under it.
func (s *state) handleStats(ctx context.Context, b *bot.Bot, update *models.Update) error {
userID, ok := senderInfo(update)
if !ok {
internal/modules/stock/handlers_test.go
+4 -4
View File
@@ -243,7 +243,7 @@ func (s *countingPortfolioStore) Put(ctx context.Context, id string, p Portfolio
func seedStockPortfolio(t *testing.T, store Store, userID int64, held int64, balance float64) {
t.Helper()
p := NewPortfolio(123)
p.Assets["TCB"] = AssetPosition{Quantity: held, Base: float64(held) * 30_000, DividendCheckedAt: 100}
p.Assets["TCB"] = AssetPosition{Quantity: held, Base: float64(held) * 30_000, OpenedAt: 100}
p.VND = balance
if err := SavePortfolio(context.Background(), store, userID, p); err != nil {
t.Fatalf("seed portfolio: %v", err)
@@ -269,7 +269,7 @@ func TestHandleCashDividendAllowsRepeatedManualAdjustments(t *testing.T) {
if got, want := p.VND, float64(418000); got != want {
t.Fatalf("balance = %v, want %v", got, want)
}
if p.Assets["TCB"].Base != 139*30_000 || p.Assets["TCB"].DividendCheckedAt != 123 {
if p.Assets["TCB"].Base != 139*30_000 || p.Assets["TCB"].OpenedAt != 100 {
t.Fatalf("cash dividend position: %+v", p.Assets["TCB"])
}
}
@@ -309,7 +309,7 @@ func TestHandleShareDividendPreservesRatioAndFloors(t *testing.T) {
if got, want := p.Assets["TCB"].Quantity, int64(152); got != want {
t.Fatalf("holding = %d, want %d", got, want)
}
if p.Assets["TCB"].Base != 139*30_000 || p.Assets["TCB"].DividendCheckedAt != 123 {
if p.Assets["TCB"].Base != 139*30_000 || p.Assets["TCB"].OpenedAt != 100 {
t.Fatalf("share dividend position: %+v", p.Assets["TCB"])
}
rb.AssertSentText(t, "Share dividend (100:10): +13 TCB")
@@ -378,7 +378,7 @@ func TestHandleCombinedDividendUsesPreEventHoldingAndOneSave(t *testing.T) {
t.Fatalf("store writes = %d, want 1", store.puts)
}
p, _ := LoadPortfolio(ctx, base, 7, 999)
if p.Assets["TCB"].Quantity != 152 || p.VND != 209500 || p.Assets["TCB"].Base != 139*30_000 || p.Assets["TCB"].DividendCheckedAt != 123 {
if p.Assets["TCB"].Quantity != 152 || p.VND != 209500 || p.Assets["TCB"].Base != 139*30_000 || p.Assets["TCB"].OpenedAt != 100 {
t.Fatalf("portfolio = %+v", p)
}
rb.AssertSentText(t, "Dividend for TCB (100:10)")
internal/modules/stock/pending_dividend.go
+21 -27
View File
@@ -3,11 +3,11 @@ package stock
import (
"context"
"crypto/rand"
"crypto/sha256"
"encoding/base64"
"errors"
"fmt"
"regexp"
"strings"
"time"
"github.com/tiennm99/miti99bot/internal/storage"
@@ -18,39 +18,29 @@ const (
pendingDividendPrefix = "pending-dividend:"
pendingDividendTTL = 24 * time.Hour
dividendTokenBytes = 16
dividendTokenLength = 22
)
var dividendTokenPattern = regexp.MustCompile(`^[A-Za-z0-9_-]{22}$`)
type PendingDividendStore = storage.DocStore[PendingDividendAction]
// PendingDividendAction is the trusted server-side half of an inline button.
// Telegram receives only an opaque random token; all financial values and the
// owner binding remain in storage.
// PendingDividendAction is the server-side half of an inline button. Financial
// values live in the user's dividend history; Telegram receives only an opaque
// random token.
type PendingDividendAction struct {
OwnerUserID int64 `json:"ownerUserId" bson:"ownerUserId"`
ChatID int64 `json:"chatId" bson:"chatId"`
MessageID int `json:"messageId" bson:"messageId"`
ProviderEventID string `json:"providerEventId" bson:"providerEventId"`
Symbol string `json:"symbol" bson:"symbol"`
Kind DividendKind `json:"kind" bson:"kind"`
VNDPerShare int64 `json:"vndPerShare,omitempty" bson:"vndPerShare,omitempty"`
OwnedShares int64 `json:"ownedShares,omitempty" bson:"ownedShares,omitempty"`
NewShares int64 `json:"newShares,omitempty" bson:"newShares,omitempty"`
ObservedHolding int64 `json:"observedHolding" bson:"observedHolding"`
PositionOpenedAt int64 `json:"positionOpenedAt,omitempty" bson:"positionOpenedAt,omitempty"`
CheckThrough int64 `json:"checkThrough" bson:"checkThrough"`
CreatedAt int64 `json:"createdAt" bson:"createdAt"`
ExpiresAt int64 `json:"expiresAt" bson:"expiresAt"`
OwnerUserID int64 `json:"ownerUserId" bson:"ownerUserId"`
ChatID int64 `json:"chatId" bson:"chatId"`
MessageID int `json:"messageId" bson:"messageId"`
ProviderEventID string `json:"providerEventId" bson:"providerEventId"`
Symbol string `json:"symbol" bson:"symbol"`
PositionOpenedAt int64 `json:"positionOpenedAt,omitempty" bson:"positionOpenedAt,omitempty"`
CreatedAt int64 `json:"createdAt" bson:"createdAt"`
ExpiresAt int64 `json:"expiresAt" bson:"expiresAt"`
}
func pendingDividendKey(token string) string { return pendingDividendPrefix + token }
func dividendLedgerKey(providerID string) string {
sum := sha256.Sum256([]byte("ssi:" + providerID))
return "ssi:" + base64.RawURLEncoding.EncodeToString(sum[:16])
}
func generateDividendToken() (string, error) {
raw := make([]byte, dividendTokenBytes)
if _, err := rand.Read(raw); err != nil {
@@ -59,12 +49,16 @@ func generateDividendToken() (string, error) {
return base64.RawURLEncoding.EncodeToString(raw), nil
}
func validDividendToken(token string) bool {
return len(token) == dividendTokenLength && dividendTokenPattern.MatchString(token)
}
func callbackToken(data string) (string, bool) {
if len(data) != len(dividendCallbackPrefix)+22 || data[:len(dividendCallbackPrefix)] != dividendCallbackPrefix {
token, ok := strings.CutPrefix(data, dividendCallbackPrefix)
if !ok || !validDividendToken(token) {
return "", false
}
token := data[len(dividendCallbackPrefix):]
return token, dividendTokenPattern.MatchString(token)
return token, true
}
func (s *state) createPendingDividend(ctx context.Context, action PendingDividendAction) (string, error) {
@@ -80,7 +74,7 @@ func (s *state) createPendingDividend(ctx context.Context, action PendingDividen
if err != nil {
return "", err
}
if !dividendTokenPattern.MatchString(token) {
if !validDividendToken(token) {
return "", errors.New("stock: invalid generated dividend token")
}
err = s.pending.PutVersioned(ctx, pendingDividendKey(token), 0, action)
internal/modules/stock/portfolio.go
+49 -26
View File
@@ -14,21 +14,38 @@ type Store = storage.DocStore[Portfolio]
const CollectionName = "stock"
// AssetPosition keeps the complete persisted state for one stock ticker.
// Base is total remaining VND cost, not average price. DividendCheckedAt is
// the cursor for future dividend-event discovery.
// AssetPosition keeps the complete persisted state for one open stock ticker.
// Base is total remaining VND cost, not average price.
type AssetPosition struct {
Quantity int64 `json:"quantity" bson:"quantity"`
Base float64 `json:"base" bson:"base"`
DividendCheckedAt int64 `json:"dividendCheckedAt" bson:"dividendCheckedAt"`
OpenedAt int64 `json:"openedAt,omitempty" bson:"openedAt,omitempty"`
Quantity int64 `json:"quantity" bson:"quantity"`
Base float64 `json:"base" bson:"base"`
OpenedAt int64 `json:"openedAt,omitempty" bson:"openedAt,omitempty"`
}
// DividendRecord is one normalized SSI event in a user's retained history.
// The raw SSI event ID is the containing map key.
type DividendRecord struct {
Kind DividendKind `json:"kind" bson:"kind"`
PublishedAt int64 `json:"publishedAt" bson:"publishedAt"`
ExDate int64 `json:"exDate,omitempty" bson:"exDate,omitempty"`
RecordDate int64 `json:"recordDate,omitempty" bson:"recordDate,omitempty"`
PaymentDate int64 `json:"paymentDate,omitempty" bson:"paymentDate,omitempty"`
VNDPerShare int64 `json:"vndPerShare,omitempty" bson:"vndPerShare,omitempty"`
OwnedShares int64 `json:"ownedShares,omitempty" bson:"ownedShares,omitempty"`
NewShares int64 `json:"newShares,omitempty" bson:"newShares,omitempty"`
Title string `json:"title,omitempty" bson:"title,omitempty"`
SourceURL string `json:"sourceUrl,omitempty" bson:"sourceUrl,omitempty"`
Processed bool `json:"processed" bson:"processed"`
}
type Portfolio struct {
VND float64 `json:"vnd" bson:"vnd"`
Assets map[string]AssetPosition `json:"assets" bson:"assets"`
AppliedDividendEvents map[string]int64 `json:"appliedDividendEvents,omitempty" bson:"appliedDividendEvents,omitempty"`
Meta PortfolioMeta `json:"meta" bson:"meta"`
VND float64 `json:"vnd" bson:"vnd"`
Assets map[string]AssetPosition `json:"assets" bson:"assets"`
Dividends map[string]map[string]DividendRecord `json:"dividends,omitempty" bson:"dividends,omitempty"`
Meta PortfolioMeta `json:"meta" bson:"meta"`
}
type PortfolioMeta struct {
@@ -37,7 +54,11 @@ type PortfolioMeta struct {
}
func NewPortfolio(now int64) Portfolio {
return Portfolio{Assets: map[string]AssetPosition{}, Meta: PortfolioMeta{CreatedAt: now}}
return Portfolio{
Assets: map[string]AssetPosition{},
Dividends: map[string]map[string]DividendRecord{},
Meta: PortfolioMeta{CreatedAt: now},
}
}
func portfolioKey(userID int64) string {
@@ -51,8 +72,8 @@ func LoadPortfolio(ctx context.Context, store Store, userID int64, now int64) (P
if p.Assets == nil {
p.Assets = map[string]AssetPosition{}
}
if p.AppliedDividendEvents == nil {
p.AppliedDividendEvents = map[string]int64{}
if p.Dividends == nil {
p.Dividends = map[string]map[string]DividendRecord{}
}
if p.Meta.CreatedAt == 0 {
p.Meta.CreatedAt = now
@@ -87,13 +108,19 @@ func (p Portfolio) Validate() error {
if err != nil || canonical != symbol {
return fmt.Errorf("stock: invalid ticker %q", symbol)
}
if position.Quantity <= 0 || !isPositiveFiniteCost(position.Base) || position.DividendCheckedAt <= 0 || position.OpenedAt < 0 {
if position.Quantity <= 0 || !isPositiveFiniteCost(position.Base) || position.OpenedAt < 0 {
return fmt.Errorf("stock: %s has invalid position", symbol)
}
}
for eventID, appliedAt := range p.AppliedDividendEvents {
if eventID == "" || len(eventID) > 128 || appliedAt <= 0 {
return fmt.Errorf("stock: invalid applied dividend event")
for symbol, events := range p.Dividends {
canonical, err := normalizeStockSymbol(symbol)
if err != nil || canonical != symbol || events == nil {
return fmt.Errorf("stock: invalid dividend ticker %q", symbol)
}
for eventID, event := range events {
if !ssiProviderIDPattern.MatchString(eventID) || !event.valid() {
return fmt.Errorf("stock: invalid dividend event %q for %s", eventID, symbol)
}
}
}
return nil
@@ -111,8 +138,8 @@ func (p *Portfolio) DeductVND(amount float64) (ok bool, balance float64) {
return true, p.VND
}
// BuyTicker adds quantity and basis. DividendCheckedAt is set only when opening
// a new position so later buys cannot skip events after the original buy.
// BuyTicker adds quantity and basis. OpenedAt identifies the current position
// lifecycle and is reset only after a full exit.
func (p *Portfolio) BuyTicker(symbol string, quantity int64, base float64, now int64) error {
if quantity <= 0 || !isPositiveFiniteCost(base) || now <= 0 {
return fmt.Errorf("stock: invalid purchase position")
@@ -130,9 +157,6 @@ func (p *Portfolio) BuyTicker(symbol string, quantity int64, base float64, now i
if !isPositiveFiniteCost(position.Base) {
return fmt.Errorf("stock: cost basis overflows")
}
if position.DividendCheckedAt == 0 {
position.DividendCheckedAt = now
}
if isOpening {
position.OpenedAt = now
}
@@ -140,8 +164,8 @@ func (p *Portfolio) BuyTicker(symbol string, quantity int64, base float64, now i
return nil
}
// SellTicker removes proportional weighted-average basis while preserving the
// dividend cursor. A full exit removes the entire ticker document.
// SellTicker removes proportional weighted-average basis. A full exit removes
// the active position but retained dividend history remains separate.
func (p *Portfolio) SellTicker(symbol string, quantity int64) (remaining int64, soldBase float64, ok bool, err error) {
position, exists := p.Assets[symbol]
if !exists || position.Quantity < quantity || quantity <= 0 {
@@ -170,7 +194,6 @@ func (p *Portfolio) ApplyDividend(symbol string, quantity int64, vnd float64, no
return fmt.Errorf("stock: invalid dividend position")
}
position.Quantity = quantity
position.DividendCheckedAt = now
p.Assets[symbol] = position
p.VND = vnd
return nil
internal/modules/stock/portfolio_test.go
+30 -8
View File
@@ -16,7 +16,7 @@ func TestLoadPortfolioFirstTimeUser(t *testing.T) {
if err != nil {
t.Fatal(err)
}
if p.VND != 0 || p.Assets == nil || p.Meta.CreatedAt != 1234567890 {
if p.VND != 0 || p.Assets == nil || p.Dividends == nil || p.Meta.CreatedAt != 1234567890 {
t.Fatalf("portfolio=%+v", p)
}
}
@@ -30,6 +30,9 @@ func TestSaveAndLoadRoundTrip(t *testing.T) {
t.Fatal(err)
}
p.Meta.Invested = 5_000_000
p.Dividends["TCB"] = map[string]DividendRecord{
"2612974": {Kind: DividendKindCash, PublishedAt: 2, RecordDate: 3, VNDPerShare: 1500},
}
if err := SavePortfolio(ctx, store, 42, p); err != nil {
t.Fatal(err)
}
@@ -38,12 +41,12 @@ func TestSaveAndLoadRoundTrip(t *testing.T) {
t.Fatal(err)
}
position := got.Assets["TCB"]
if got.VND != 5_000_000 || position.Quantity != 100 || position.Base != 3_000_000 || position.DividendCheckedAt != 10 || got.Meta.CreatedAt != 1 {
if got.VND != 5_000_000 || position.Quantity != 100 || position.Base != 3_000_000 || position.OpenedAt != 10 || got.Meta.CreatedAt != 1 || got.Dividends["TCB"]["2612974"].VNDPerShare != 1500 {
t.Fatalf("portfolio=%+v", got)
}
}
func TestBuyPreservesDividendCheckedAtAndSellUsesWeightedBasis(t *testing.T) {
func TestBuyPreservesOpenedAtAndSellUsesWeightedBasis(t *testing.T) {
p := NewPortfolio(1)
if err := p.BuyTicker("TCB", 100, 2_000_000, 10); err != nil {
t.Fatal(err)
@@ -52,27 +55,46 @@ func TestBuyPreservesDividendCheckedAtAndSellUsesWeightedBasis(t *testing.T) {
t.Fatal(err)
}
position := p.Assets["TCB"]
if position.DividendCheckedAt != 10 {
t.Fatalf("additional buy changed dividend cursor to %d", position.DividendCheckedAt)
if position.OpenedAt != 10 {
t.Fatalf("additional buy changed openedAt to %d", position.OpenedAt)
}
remaining, soldBase, ok, err := p.SellTicker("TCB", 60)
if err != nil || !ok || remaining != 90 || soldBase != 1_400_000 {
t.Fatalf("remaining=%d soldBase=%v ok=%v err=%v", remaining, soldBase, ok, err)
}
position = p.Assets["TCB"]
if position.Base != 2_100_000 || position.DividendCheckedAt != 10 {
if position.Base != 2_100_000 || position.OpenedAt != 10 {
t.Fatalf("position=%+v", position)
}
}
func TestDividendAdvancesCursorWithoutChangingBase(t *testing.T) {
func TestDividendDoesNotChangeLifecycleOrBase(t *testing.T) {
p := NewPortfolio(1)
_ = p.BuyTicker("TCB", 100, 3_000_000, 10)
if err := p.ApplyDividend("TCB", 110, 500_000, 30); err != nil {
t.Fatal(err)
}
position := p.Assets["TCB"]
if position.Quantity != 110 || position.Base != 3_000_000 || position.DividendCheckedAt != 30 || p.VND != 500_000 {
if position.Quantity != 110 || position.Base != 3_000_000 || position.OpenedAt != 10 || p.VND != 500_000 {
t.Fatalf("portfolio=%+v", p)
}
}
func TestFullSaleKeepsDividendHistorySeparate(t *testing.T) {
p := NewPortfolio(1)
if err := p.BuyTicker("TCB", 100, 3_000_000, 10); err != nil {
t.Fatal(err)
}
p.Dividends["TCB"] = map[string]DividendRecord{
"2612974": {Kind: DividendKindCash, PublishedAt: 2, VNDPerShare: 1500, Processed: true},
}
if _, _, ok, err := p.SellTicker("TCB", 100); err != nil || !ok {
t.Fatalf("full sale: ok=%v err=%v", ok, err)
}
if _, ok := p.Assets["TCB"]; ok {
t.Fatal("full sale retained active asset")
}
if !p.Dividends["TCB"]["2612974"].Processed {
t.Fatal("full sale removed dividend history")
}
}
internal/modules/stock/startup.go
+148
View File
@@ -0,0 +1,148 @@
package stock
import (
"context"
"errors"
"fmt"
"time"
"github.com/tiennm99/miti99bot/internal/log"
"github.com/tiennm99/miti99bot/internal/storage"
"github.com/tiennm99/miti99bot/internal/systemstate"
)
const (
dividendHistoryMigrationMarkerKey = "migration:stock-dividend-history-v1"
dividendHistoryMigrationRetries = 5
)
// legacyDividendAssetPosition contains the retired discovery cursor so a
// whole-document replacement can physically remove it from MongoDB.
type legacyDividendAssetPosition struct {
Quantity int64 `json:"quantity" bson:"quantity"`
Base float64 `json:"base" bson:"base"`
DividendCheckedAt *int64 `json:"dividendCheckedAt,omitempty" bson:"dividendCheckedAt,omitempty"`
OpenedAt int64 `json:"openedAt,omitempty" bson:"openedAt,omitempty"`
}
// legacyDividendPortfolio mirrors both the current portfolio fields and the
// retired applied-event ledger. Keeping Dividends here is important: migration
// must never discard history already written by the new runtime.
type legacyDividendPortfolio struct {
VND float64 `json:"vnd" bson:"vnd"`
Assets map[string]legacyDividendAssetPosition `json:"assets" bson:"assets"`
Dividends map[string]map[string]DividendRecord `json:"dividends,omitempty" bson:"dividends,omitempty"`
AppliedDividendEvents map[string]int64 `json:"appliedDividendEvents,omitempty" bson:"appliedDividendEvents,omitempty"`
Meta PortfolioMeta `json:"meta" bson:"meta"`
}
// InitStore removes the cursor-era dividend fields from stock user portfolios.
// The completion marker makes the one-time migration idempotent; it is written
// only after every listed user document has been handled successfully.
func InitStore(ctx context.Context, portfolioColl, systemColl storage.Collection) error {
system := systemstate.New(systemColl)
marker, exists, err := system.Get(ctx, dividendHistoryMigrationMarkerKey)
if err != nil {
return fmt.Errorf("stock dividend history migration: read marker: %w", err)
}
if exists && marker.Status == "completed" {
return nil
}
docs := storage.Typed[legacyDividendPortfolio](portfolioColl)
keys, err := docs.List(ctx, "user:")
if err != nil {
return fmt.Errorf("stock dividend history migration: list portfolios: %w", err)
}
var migrated int64
for index, key := range keys {
changed, err := migrateDividendHistorySchema(ctx, docs, key)
if err != nil {
return err
}
if changed {
migrated++
log.Info("stock dividend history migrated", "portfolio", index+1, "total", len(keys))
}
}
marker = completedDividendHistoryMigration(marker, exists, migrated, time.Now().UnixMilli())
if err := system.Put(ctx, dividendHistoryMigrationMarkerKey, marker); err != nil {
return fmt.Errorf("stock dividend history migration: write marker: %w", err)
}
return nil
}
func completedDividendHistoryMigration(marker systemstate.Record, exists bool, migrated, now int64) systemstate.Record {
if !exists {
marker = systemstate.Record{
Kind: "migration",
Name: "stock dividend history v1",
}
}
if marker.CompletedAt == 0 {
marker.CompletedAt = now
}
marker.Status = "completed"
marker.Count += migrated
marker.UpdatedAt = now
return marker
}
func migrateDividendHistorySchema(ctx context.Context, docs storage.DocStore[legacyDividendPortfolio], key string) (bool, error) {
for attempt := 0; attempt < dividendHistoryMigrationRetries; attempt++ {
doc, version, err := docs.Get(ctx, key)
if err != nil {
return false, fmt.Errorf("stock dividend history migration: read %s: %w", key, err)
}
if !doc.hasLegacyDividendFields() {
return false, nil
}
current := doc.currentPortfolio()
if err := current.Validate(); err != nil {
return false, fmt.Errorf("stock dividend history migration: validate %s: %w", key, err)
}
err = docs.PutVersioned(ctx, key, version, doc.withoutLegacyDividendFields())
if err == nil {
return true, nil
}
if !errors.Is(err, storage.ErrConflict) {
return false, fmt.Errorf("stock dividend history migration: write %s: %w", key, err)
}
}
return false, fmt.Errorf("stock dividend history migration: write %s: %w", key, storage.ErrConflict)
}
func (p legacyDividendPortfolio) hasLegacyDividendFields() bool {
if p.AppliedDividendEvents != nil {
return true
}
for _, position := range p.Assets {
if position.DividendCheckedAt != nil {
return true
}
}
return false
}
func (p legacyDividendPortfolio) withoutLegacyDividendFields() legacyDividendPortfolio {
p.AppliedDividendEvents = nil
for ticker, position := range p.Assets {
position.DividendCheckedAt = nil
p.Assets[ticker] = position
}
return p
}
func (p legacyDividendPortfolio) currentPortfolio() Portfolio {
assets := make(map[string]AssetPosition, len(p.Assets))
for ticker, position := range p.Assets {
assets[ticker] = AssetPosition{
Quantity: position.Quantity,
Base: position.Base,
OpenedAt: position.OpenedAt,
}
}
return Portfolio{VND: p.VND, Assets: assets, Dividends: p.Dividends, Meta: p.Meta}
}
internal/modules/stock/startup_mongo_test.go
+88
View File
@@ -0,0 +1,88 @@
package stock
import (
"context"
"fmt"
"os"
"testing"
"time"
"go.mongodb.org/mongo-driver/v2/bson"
"github.com/tiennm99/miti99bot/internal/storage"
"github.com/tiennm99/miti99bot/internal/systemstate"
"github.com/tiennm99/miti99bot/internal/testutil/mongotest"
)
var stockMongoTests mongotest.Manager
func TestMain(m *testing.M) {
os.Exit(stockMongoTests.Run(m))
}
func TestInitStorePhysicallyDropsLegacyDividendFieldsInMongoDB(t *testing.T) {
ctx, portfolioColl, systemColl := setupMongoStockMigrationTest(t)
docs := storage.Typed[legacyDividendPortfolio](portfolioColl)
if err := docs.Put(ctx, "user:7", legacyDividendPortfolio{
VND: 50_000,
Assets: map[string]legacyDividendAssetPosition{
"TCB": {Quantity: 100, Base: 3_000_000, DividendCheckedAt: legacyCursor(123), OpenedAt: 99},
},
Dividends: map[string]map[string]DividendRecord{
"TCB": {"2612974": legacyPreservedDividend()},
},
AppliedDividendEvents: map[string]int64{"old-hash": 456},
Meta: PortfolioMeta{Invested: 3_000_000, CreatedAt: 1},
}); err != nil {
t.Fatal(err)
}
if err := InitStore(ctx, portfolioColl, systemColl); err != nil {
t.Fatalf("InitStore: %v", err)
}
if err := InitStore(ctx, portfolioColl, systemColl); err != nil {
t.Fatalf("second InitStore: %v", err)
}
rawColl, ok := storage.MongoCollection(portfolioColl)
if !ok {
t.Fatal("portfolio collection is not MongoDB-backed")
}
var raw bson.Raw
if err := rawColl.FindOne(ctx, bson.M{"_id": "user:7"}).Decode(&raw); err != nil {
t.Fatal(err)
}
if _, err := raw.LookupErr("appliedDividendEvents"); err == nil {
t.Fatalf("root legacy ledger remains in %v", raw)
}
position := raw.Lookup("assets").Document().Lookup("TCB").Document()
if _, err := position.LookupErr("dividendCheckedAt"); err == nil {
t.Fatalf("asset legacy cursor remains in %v", position)
}
if position.Lookup("openedAt").Int64() != 99 || raw.Lookup("vnd").Double() != 50_000 {
t.Fatalf("current portfolio fields changed: %v", raw)
}
if _, err := raw.LookupErr("dividends"); err != nil {
t.Fatalf("new dividend history was discarded: %v", raw)
}
}
func setupMongoStockMigrationTest(t *testing.T) (context.Context, storage.Collection, storage.Collection) {
t.Helper()
uri := stockMongoTests.URI(t)
ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
t.Cleanup(cancel)
client, err := storage.NewMongoClient(ctx, uri)
if err != nil {
t.Fatal(err)
}
db := client.Database(fmt.Sprintf("miti99bot_stock_migration_test_%d", time.Now().UnixNano()))
t.Cleanup(func() {
cleanupCtx, cleanupCancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cleanupCancel()
_ = db.Drop(cleanupCtx)
_ = client.Disconnect(cleanupCtx)
})
provider := storage.NewMongoProvider(db)
return ctx, provider.Collection(CollectionName), provider.Collection(systemstate.CollectionName)
}
internal/modules/stock/startup_test.go
+165
View File
@@ -0,0 +1,165 @@
package stock
import (
"context"
"errors"
"testing"
"github.com/tiennm99/miti99bot/internal/storage"
"github.com/tiennm99/miti99bot/internal/systemstate"
)
func legacyCursor(value int64) *int64 { return &value }
func legacyPreservedDividend() DividendRecord {
return DividendRecord{
Kind: DividendKindCash,
PublishedAt: 1,
RecordDate: 2,
VNDPerShare: 1_500,
Title: "Cash dividend",
SourceURL: "https://example.test/event/2612974",
}
}
func TestInitStoreRemovesLegacyDividendFields(t *testing.T) {
ctx := context.Background()
provider := storage.NewMemoryProvider()
portfolioColl := provider.Collection(CollectionName)
systemColl := provider.Collection(systemstate.CollectionName)
docs := storage.Typed[legacyDividendPortfolio](portfolioColl)
legacy := legacyDividendPortfolio{
VND: 50_000,
Assets: map[string]legacyDividendAssetPosition{
"TCB": {Quantity: 100, Base: 3_000_000, DividendCheckedAt: legacyCursor(123), OpenedAt: 99},
},
Dividends: map[string]map[string]DividendRecord{
"TCB": {"2612974": legacyPreservedDividend()},
},
AppliedDividendEvents: map[string]int64{"old-hash": 456},
Meta: PortfolioMeta{Invested: 3_000_000, CreatedAt: 1},
}
if err := docs.Put(ctx, "user:7", legacy); err != nil {
t.Fatal(err)
}
if err := docs.Put(ctx, "pending-dividend:leave-me", legacy); err != nil {
t.Fatal(err)
}
if err := InitStore(ctx, portfolioColl, systemColl); err != nil {
t.Fatalf("InitStore: %v", err)
}
got, _, err := docs.Get(ctx, "user:7")
if err != nil {
t.Fatal(err)
}
if got.AppliedDividendEvents != nil || got.Assets["TCB"].DividendCheckedAt != nil {
t.Fatalf("legacy fields remain: %+v", got)
}
if got.VND != legacy.VND || got.Assets["TCB"].Quantity != 100 || got.Assets["TCB"].Base != 3_000_000 || got.Assets["TCB"].OpenedAt != 99 || got.Meta != legacy.Meta {
t.Fatalf("portfolio data changed: got=%+v want=%+v", got, legacy)
}
if event, ok := got.Dividends["TCB"]["2612974"]; !ok || event != legacyPreservedDividend() {
t.Fatalf("dividend history was discarded: %+v", got.Dividends)
}
pending, _, err := docs.Get(ctx, "pending-dividend:leave-me")
if err != nil {
t.Fatal(err)
}
if pending.AppliedDividendEvents == nil || pending.Assets["TCB"].DividendCheckedAt == nil {
t.Fatalf("non-user document was rewritten: %+v", pending)
}
marker, exists, err := systemstate.New(systemColl).Get(ctx, dividendHistoryMigrationMarkerKey)
if err != nil || !exists || marker.Status != "completed" || marker.Count != 1 {
t.Fatalf("marker=%+v exists=%v err=%v", marker, exists, err)
}
if err := InitStore(ctx, portfolioColl, systemColl); err != nil {
t.Fatalf("second InitStore: %v", err)
}
marker, _, _ = systemstate.New(systemColl).Get(ctx, dividendHistoryMigrationMarkerKey)
if marker.Count != 1 {
t.Fatalf("idempotent marker count=%d, want 1", marker.Count)
}
}
type conflictOnceDividendMigrationStore struct {
storage.DocStore[legacyDividendPortfolio]
conflicted bool
}
func (s *conflictOnceDividendMigrationStore) PutVersioned(ctx context.Context, key string, version int64, value legacyDividendPortfolio) error {
if !s.conflicted {
s.conflicted = true
return storage.ErrConflict
}
return s.DocStore.PutVersioned(ctx, key, version, value)
}
func TestDividendHistoryMigrationRetriesVersionConflict(t *testing.T) {
ctx := context.Background()
provider := storage.NewMemoryProvider()
docs := storage.Typed[legacyDividendPortfolio](provider.Collection(CollectionName))
if err := docs.Put(ctx, "user:7", legacyDividendPortfolio{
Assets: map[string]legacyDividendAssetPosition{
"TCB": {Quantity: 10, Base: 300_000, DividendCheckedAt: legacyCursor(123), OpenedAt: 99},
},
}); err != nil {
t.Fatal(err)
}
store := &conflictOnceDividendMigrationStore{DocStore: docs}
changed, err := migrateDividendHistorySchema(ctx, store, "user:7")
if err != nil || !changed || !store.conflicted {
t.Fatalf("changed=%v conflicted=%v err=%v", changed, store.conflicted, err)
}
}
type alwaysConflictDividendMigrationStore struct {
storage.DocStore[legacyDividendPortfolio]
}
func (s alwaysConflictDividendMigrationStore) PutVersioned(context.Context, string, int64, legacyDividendPortfolio) error {
return storage.ErrConflict
}
func TestDividendHistoryMigrationReturnsExhaustedConflict(t *testing.T) {
ctx := context.Background()
provider := storage.NewMemoryProvider()
docs := storage.Typed[legacyDividendPortfolio](provider.Collection(CollectionName))
if err := docs.Put(ctx, "user:7", legacyDividendPortfolio{
Assets: map[string]legacyDividendAssetPosition{
"TCB": {Quantity: 10, Base: 300_000, DividendCheckedAt: legacyCursor(123), OpenedAt: 99},
},
}); err != nil {
t.Fatal(err)
}
changed, err := migrateDividendHistorySchema(ctx, alwaysConflictDividendMigrationStore{docs}, "user:7")
if changed || !errors.Is(err, storage.ErrConflict) {
t.Fatalf("changed=%v err=%v, want wrapped conflict", changed, err)
}
}
func TestInitStoreDoesNotMarkFailedMigrationComplete(t *testing.T) {
ctx := context.Background()
provider := storage.NewMemoryProvider()
// A legacy portfolio with an invalid position must fail validation before
// its retired fields are removed.
if err := storage.Typed[legacyDividendPortfolio](provider.Collection(CollectionName)).Put(ctx, "user:7", legacyDividendPortfolio{
Assets: map[string]legacyDividendAssetPosition{
"TCB": {Quantity: 10, Base: 0, DividendCheckedAt: legacyCursor(123), OpenedAt: 99},
},
}); err != nil {
t.Fatal(err)
}
if err := InitStore(ctx, provider.Collection(CollectionName), provider.Collection(systemstate.CollectionName)); err == nil {
t.Fatal("InitStore accepted invalid legacy portfolio")
}
_, exists, err := systemstate.New(provider.Collection(systemstate.CollectionName)).Get(ctx, dividendHistoryMigrationMarkerKey)
if err != nil || exists {
t.Fatalf("completion marker exists=%v after failure, err=%v", exists, err)
}
}