diff --git a/internal/modules/dispatcher.go b/internal/modules/dispatcher.go
index aa9b606..373af35 100644
--- a/internal/modules/dispatcher.go
+++ b/internal/modules/dispatcher.go
@@ -28,10 +28,18 @@ func (a Auth) Permits(v Visibility, update *models.Update) bool {
if v == VisibilityPublic {
return true
}
- if update == nil || update.Message == nil || update.Message.From == nil {
+ if update == nil {
+ return false
+ }
+ var senderID int64
+ if update.Message != nil && update.Message.From != nil {
+ senderID = update.Message.From.ID
+ } else if update.CallbackQuery != nil {
+ senderID = update.CallbackQuery.From.ID
+ }
+ if senderID == 0 {
return false
}
- senderID := update.Message.From.ID
switch v {
case VisibilityPrivate:
return a.BotOwnerID != 0 && senderID == a.BotOwnerID
@@ -82,6 +90,24 @@ func Install(b *bot.Bot, reg *Registry, auth Auth) {
},
)
}
+ for prefix, callback := range reg.callbacks {
+ callbackCopy := callback
+ prefixCopy := prefix
+ b.RegisterHandler(bot.HandlerTypeCallbackQueryData, prefixCopy, bot.MatchTypePrefix,
+ func(ctx context.Context, b *bot.Bot, update *models.Update) {
+ if !auth.Permits(callbackCopy.Visibility, update) {
+ if update != nil && update.CallbackQuery != nil {
+ _, _ = b.AnswerCallbackQuery(ctx, &bot.AnswerCallbackQueryParams{CallbackQueryID: update.CallbackQuery.ID})
+ }
+ return
+ }
+ err := callbackCopy.Handler(ctx, b, update)
+ if err != nil {
+ metrics.IncError("callback-handler-error")
+ log.Error("callback", "prefix", prefixCopy, "err", err)
+ }
+ })
+ }
}
// logCommand emits one structured line per authorized command invocation: what
diff --git a/internal/modules/dispatcher_test.go b/internal/modules/dispatcher_test.go
index 70ca8b0..35b0963 100644
--- a/internal/modules/dispatcher_test.go
+++ b/internal/modules/dispatcher_test.go
@@ -25,6 +25,9 @@ func TestAuth_Permits(t *testing.T) {
updateFrom := func(id int64) *models.Update {
return &models.Update{Message: &models.Message{From: &models.User{ID: id}}}
}
+ callbackFrom := func(id int64) *models.Update {
+ return &models.Update{CallbackQuery: &models.CallbackQuery{From: models.User{ID: id}}}
+ }
cases := []struct {
name string
@@ -35,6 +38,7 @@ func TestAuth_Permits(t *testing.T) {
{"public-no-message", VisibilityPublic, &models.Update{}, true},
{"public-stranger", VisibilityPublic, updateFrom(stranger), true},
{"protected-owner", VisibilityProtected, updateFrom(owner), true},
+ {"protected-callback-owner", VisibilityProtected, callbackFrom(owner), true},
{"protected-admin", VisibilityProtected, updateFrom(admin), true},
{"protected-stranger", VisibilityProtected, updateFrom(stranger), false},
{"private-owner", VisibilityPrivate, updateFrom(owner), true},
diff --git a/internal/modules/module.go b/internal/modules/module.go
index e335887..7b2d335 100644
--- a/internal/modules/module.go
+++ b/internal/modules/module.go
@@ -27,6 +27,18 @@ const (
// purely for logging/metrics, not flow control.
type CommandHandler func(ctx context.Context, b *bot.Bot, update *models.Update) error
+// CallbackHandler runs in response to inline-keyboard callback data. Callback
+// payloads are not commands and therefore do not participate in command stats.
+type CallbackHandler func(ctx context.Context, b *bot.Bot, update *models.Update) error
+
+// Callback registers an inline-keyboard callback-data prefix owned by a module.
+// Prefixes must be globally non-overlapping so one callback reaches one owner.
+type Callback struct {
+ Prefix string
+ Visibility Visibility
+ Handler CallbackHandler
+}
+
// CronHandler runs when a cron fires — driven by the in-process scheduler
// (internal/cron) on self-host, or by a POST to /cron/{name} for manual
// triggers. Crons receive the per-module-prefixed Deps via the registry;
@@ -59,6 +71,7 @@ type Cron struct {
type Module struct {
Name string
Commands []Command
+ Callbacks []Callback
Crons []Cron
CommandHook func(ctx context.Context, name string, update *models.Update) // optional; called by dispatcher after each authorized command invocation. update carries the originating Telegram update so hooks can attribute usage to a user.
}
diff --git a/internal/modules/registry.go b/internal/modules/registry.go
index d940e33..fbd125a 100644
--- a/internal/modules/registry.go
+++ b/internal/modules/registry.go
@@ -5,6 +5,7 @@ import (
"fmt"
"regexp"
"sort"
+ "strings"
"github.com/go-telegram/bot"
"github.com/go-telegram/bot/models"
@@ -36,6 +37,7 @@ type Registry struct {
private map[string]Command
crons map[string]Cron // name → Cron, unique across modules
cronDeps map[string]Deps // cron name → owning module's prefixed Deps
+ callbacks map[string]Callback
commandHooks []func(ctx context.Context, name string, update *models.Update)
}
@@ -121,10 +123,12 @@ func Build(enabled []string, factories map[string]Factory, provider storage.Prov
private: map[string]Command{},
crons: map[string]Cron{},
cronDeps: map[string]Deps{},
+ callbacks: map[string]Callback{},
}
owners := map[string]string{} // command name → module that registered it
cronOwners := map[string]string{}
+ callbackOwners := map[string]string{}
seenModule := map[string]bool{}
var unknown []string
@@ -192,6 +196,10 @@ func Build(enabled []string, factories map[string]Factory, provider storage.Prov
reg.cronDeps[cron.Name] = moduleDeps
}
+ if err := reg.addCallbacks(name, mod.Callbacks, callbackOwners); err != nil {
+ return nil, err
+ }
+
reg.Modules = append(reg.Modules, mod)
}
@@ -202,6 +210,22 @@ func Build(enabled []string, factories map[string]Factory, provider storage.Prov
return reg, nil
}
+func (r *Registry) addCallbacks(module string, callbacks []Callback, owners map[string]string) error {
+ for _, callback := range callbacks {
+ if err := validateCallback(callback); err != nil {
+ return fmt.Errorf("module %q: %w", module, err)
+ }
+ for prefix, owner := range owners {
+ if strings.HasPrefix(prefix, callback.Prefix) || strings.HasPrefix(callback.Prefix, prefix) {
+ return fmt.Errorf("callback prefix conflict: %q in %q overlaps %q in %q", callback.Prefix, module, prefix, owner)
+ }
+ }
+ owners[callback.Prefix] = module
+ r.callbacks[callback.Prefix] = callback
+ }
+ return nil
+}
+
func sortedCommands(m map[string]Command) []Command {
out := make([]Command, 0, len(m))
for _, c := range m {
diff --git a/internal/modules/registry_test.go b/internal/modules/registry_test.go
index a594abf..539b2f4 100644
--- a/internal/modules/registry_test.go
+++ b/internal/modules/registry_test.go
@@ -134,6 +134,30 @@ func TestBuild_DetectsCronConflict(t *testing.T) {
}
}
+func TestBuild_DetectsOverlappingCallbackPrefixes(t *testing.T) {
+ callback := func(prefix string) Callback {
+ return Callback{Prefix: prefix, Visibility: VisibilityPublic, Handler: func(_ context.Context, _ *bot.Bot, _ *models.Update) error { return nil }}
+ }
+ factories := map[string]Factory{
+ "alpha": func(_ Deps) Module { return Module{Callbacks: []Callback{callback("stock:")}} },
+ "beta": func(_ Deps) Module { return Module{Callbacks: []Callback{callback("stock:div:")}} },
+ }
+ _, err := Build([]string{"alpha", "beta"}, factories, newProvider(), BuildOptions{})
+ if err == nil || !strings.Contains(err.Error(), "callback prefix conflict") {
+ t.Fatalf("expected callback prefix conflict, got %v", err)
+ }
+}
+
+func TestBuild_RejectsInvalidCallback(t *testing.T) {
+ factories := map[string]Factory{
+ "alpha": func(_ Deps) Module { return Module{Callbacks: []Callback{{Prefix: "", Handler: nil}}} },
+ }
+ _, err := Build([]string{"alpha"}, factories, newProvider(), BuildOptions{})
+ if err == nil || !strings.Contains(err.Error(), "callback") {
+ t.Fatalf("expected callback validation error, got %v", err)
+ }
+}
+
func TestBuild_RequiresProvider(t *testing.T) {
_, err := Build(nil, map[string]Factory{}, nil, BuildOptions{})
if err == nil {
diff --git a/internal/modules/stock/dividend_callback.go b/internal/modules/stock/dividend_callback.go
new file mode 100644
index 0000000..e0546e3
--- /dev/null
+++ b/internal/modules/stock/dividend_callback.go
@@ -0,0 +1,196 @@
+package stock
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "strconv"
+
+ "github.com/go-telegram/bot"
+ "github.com/go-telegram/bot/models"
+
+ "github.com/tiennm99/miti99bot/internal/log"
+ "github.com/tiennm99/miti99bot/internal/modules/util/chathelper"
+ "github.com/tiennm99/miti99bot/internal/storage"
+)
+
+func answerDividendCallback(ctx context.Context, b *bot.Bot, queryID, text string, alert bool) error {
+ _, err := b.AnswerCallbackQuery(ctx, &bot.AnswerCallbackQueryParams{
+ CallbackQueryID: queryID,
+ Text: text,
+ ShowAlert: alert,
+ })
+ return err
+}
+
+func removeDividendButton(ctx context.Context, b *bot.Bot, chatID int64, messageID int) error {
+ _, err := b.EditMessageReplyMarkup(ctx, &bot.EditMessageReplyMarkupParams{
+ ChatID: chatID,
+ MessageID: messageID,
+ ReplyMarkup: &models.InlineKeyboardMarkup{},
+ })
+ return err
+}
+
+func (s *state) handleDividendCallback(ctx context.Context, b *bot.Bot, update *models.Update) error {
+ if update == nil || update.CallbackQuery == nil {
+ return nil
+ }
+ query := update.CallbackQuery
+ action, msg, actionKey, handled, err := s.resolveDividendCallback(ctx, b, query)
+ if err != nil || handled {
+ return err
+ }
+ defer s.locks.Acquire(strconv.FormatInt(action.OwnerUserID, 10))()
+ return s.consumeDividendAction(ctx, b, query, msg, actionKey, action)
+}
+
+func (s *state) resolveDividendCallback(ctx context.Context, b *bot.Bot, query *models.CallbackQuery) (PendingDividendAction, *models.Message, string, bool, error) {
+ token, ok := callbackToken(query.Data)
+ if !ok || s.pending == nil {
+ err := answerDividendCallback(ctx, b, query.ID, "This dividend suggestion is invalid.", true)
+ return PendingDividendAction{}, nil, "", true, err
+ }
+ actionKey := pendingDividendKey(token)
+ action, _, err := s.pending.Get(ctx, actionKey)
+ if errors.Is(err, storage.ErrNotFound) {
+ err = answerDividendCallback(ctx, b, query.ID, "This dividend suggestion expired or was already used.", true)
+ return PendingDividendAction{}, nil, "", true, err
+ }
+ if err != nil {
+ log.Error("stock_load_dividend_action", "err", err)
+ err = answerDividendCallback(ctx, b, query.ID, "Could not load this dividend suggestion. Try again.", true)
+ return PendingDividendAction{}, nil, "", true, err
+ }
+ if query.From.ID == 0 || query.From.ID != action.OwnerUserID {
+ err = answerDividendCallback(ctx, b, query.ID, "Only the user who requested this portfolio can apply this dividend.", true)
+ return PendingDividendAction{}, nil, "", true, err
+ }
+ msg := query.Message.Message
+ if msg == nil || msg.Chat.ID != action.ChatID || msg.ID != action.MessageID || action.MessageID == 0 {
+ err = answerDividendCallback(ctx, b, query.ID, "This dividend suggestion is no longer valid here.", true)
+ return PendingDividendAction{}, nil, "", true, err
+ }
+ 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)
+ return PendingDividendAction{}, nil, "", true, err
+ }
+ return action, msg, actionKey, false, nil
+}
+
+func (s *state) consumeDividendAction(ctx context.Context, b *bot.Bot, query *models.CallbackQuery, msg *models.Message, actionKey string, action PendingDividendAction) error {
+ now := s.now().UnixMilli()
+ // Re-read under the user lock so two callback workers cannot consume the
+ // same action using stale state if handler execution becomes asynchronous.
+ action, _, err := s.pending.Get(ctx, actionKey)
+ if errors.Is(err, storage.ErrNotFound) {
+ return answerDividendCallback(ctx, b, query.ID, "This dividend was already handled.", true)
+ }
+ if err != nil {
+ return fmt.Errorf("reload pending dividend action: %w", err)
+ }
+ p, err := LoadPortfolio(ctx, s.store, action.OwnerUserID, now)
+ if err != nil {
+ 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)
+ }
+ 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)
+ }
+ 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)
+ }
+
+ result, err := applySuggestedDividend(&p, action, 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
+ 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
+ // 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)
+ }
+ 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 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)
+ }
+ 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 {
+ case DividendKindCash:
+ total, err := cashDividendTotal(held, action.VNDPerShare)
+ if err != nil {
+ return "", err
+ }
+ balance, err := checkedVNDBalance(p.VND, total)
+ if err != nil {
+ return "", err
+ }
+ if err := p.ApplyDividend(action.Symbol, held, balance, now); err != nil {
+ return "", err
+ }
+ preserveDividendCursor(p, action.Symbol, preservedCursor)
+ return "Applied cash dividend for " + action.Symbol + ": " + FormatVND(float64(action.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)}
+ newShares, err := shareDividendEntitlement(held, ratio)
+ if err != nil {
+ return "", err
+ }
+ if newShares == 0 {
+ return "", errors.New("share dividend rounds down to zero")
+ }
+ finalHolding, err := checkedHoldingAfterDividend(held, newShares)
+ if err != nil {
+ return "", err
+ }
+ if err := p.ApplyDividend(action.Symbol, finalHolding, p.VND, now); err != nil {
+ return "", err
+ }
+ preserveDividendCursor(p, action.Symbol, preservedCursor)
+ return "Applied share dividend for " + action.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
+}
diff --git a/internal/modules/stock/dividend_events.go b/internal/modules/stock/dividend_events.go
new file mode 100644
index 0000000..a00176b
--- /dev/null
+++ b/internal/modules/stock/dividend_events.go
@@ -0,0 +1,42 @@
+package stock
+
+import (
+ "context"
+ "time"
+)
+
+// DividendEventProvider returns normalized dividend events up to through. An
+// implementation may include a bounded overlap before after when its source
+// exposes only day-granularity publication times; callers must deduplicate by
+// provider event ID.
+type DividendEventProvider interface {
+ FetchDividendEvents(ctx context.Context, symbol string, after, through time.Time) ([]DividendEvent, error)
+}
+
+// DividendKind identifies how an event changes a portfolio.
+type DividendKind string
+
+const (
+ DividendKindCash DividendKind = "cash"
+ DividendKindShares DividendKind = "shares"
+)
+
+// DividendEvent is a provider-independent, validated stock dividend event.
+// Cash events set VNDPerShare. Share events set OwnedShares and NewShares.
+type DividendEvent struct {
+ ProviderID string
+ Symbol string
+ Kind DividendKind
+
+ PublishedAt time.Time
+ ExDate time.Time
+ RecordDate time.Time
+ PaymentDate time.Time
+
+ VNDPerShare int64
+ OwnedShares int64
+ NewShares int64
+
+ Title string
+ SourceURL string
+}
diff --git a/internal/modules/stock/dividend_events_ssi.go b/internal/modules/stock/dividend_events_ssi.go
new file mode 100644
index 0000000..65f11d2
--- /dev/null
+++ b/internal/modules/stock/dividend_events_ssi.go
@@ -0,0 +1,314 @@
+package stock
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "math/big"
+ "net/http"
+ "net/url"
+ "regexp"
+ "sort"
+ "strconv"
+ "strings"
+ "sync"
+ "time"
+)
+
+const (
+ ssiDividendDefaultURL = "https://iboard-api.ssi.com.vn"
+ ssiDividendPath = "/statistics/company/corporate-actions"
+ ssiDividendPageSize = 50
+ ssiDividendMaxPages = 20
+ ssiDividendTimeout = 3 * time.Second
+)
+
+var saigonLocation = time.FixedZone("Asia/Saigon", 7*60*60)
+var ssiProviderIDPattern = regexp.MustCompile(`^[A-Za-z0-9_-]{1,64}$`)
+
+// SSIDividendProvider reads corporate actions from SSI iBoard's undocumented
+// public endpoint. It is isolated behind DividendEventProvider because SSI does
+// not publish a compatibility or availability guarantee for this endpoint.
+type SSIDividendProvider struct {
+ HTTP *http.Client
+ BaseURL string
+
+ defaultOnce sync.Once
+ defaultClient *http.Client
+}
+
+type ssiDividendResponse struct {
+ Code string `json:"code"`
+ Status string `json:"status"`
+ Data []ssiDividendEvent `json:"data"`
+ Paging struct {
+ TotalPage int `json:"totalPage"`
+ Page int `json:"page"`
+ } `json:"paging"`
+}
+
+type ssiDividendEvent struct {
+ CorID string `json:"CorId"`
+ Symbol string `json:"symbol"`
+ EventListCode string `json:"eventListCode"`
+ EventName string `json:"eventName"`
+ EventTitle string `json:"eventTitle"`
+ EventDescription string `json:"eventDescription"`
+ ExrightDate string `json:"exrightDate"`
+ RecordDate string `json:"recordDate"`
+ IssueDate string `json:"issueDate"`
+ PublicDate string `json:"publicDate"`
+ Value ssiDecimal `json:"value"`
+ Ratio ssiDecimal `json:"ratio"`
+}
+
+// ssiDecimal retains the exact JSON spelling whether SSI encodes a decimal as
+// a JSON string (current behavior) or changes it to a JSON number.
+type ssiDecimal string
+
+func (d *ssiDecimal) UnmarshalJSON(data []byte) error {
+ if string(data) == "null" {
+ *d = ""
+ return nil
+ }
+ if len(data) > 0 && data[0] == '"' {
+ var value string
+ if err := json.Unmarshal(data, &value); err != nil {
+ return err
+ }
+ *d = ssiDecimal(value)
+ return nil
+ }
+ value := string(data)
+ if _, ok := new(big.Rat).SetString(value); !ok {
+ return fmt.Errorf("invalid decimal %q", value)
+ }
+ *d = ssiDecimal(value)
+ return nil
+}
+
+func (p *SSIDividendProvider) httpClient() *http.Client {
+ if p.HTTP != nil {
+ return p.HTTP
+ }
+ p.defaultOnce.Do(func() {
+ p.defaultClient = &http.Client{Timeout: ssiDividendTimeout}
+ })
+ return p.defaultClient
+}
+
+func (p *SSIDividendProvider) endpoint() string {
+ baseURL := strings.TrimRight(strings.TrimSpace(p.BaseURL), "/")
+ if baseURL == "" {
+ baseURL = ssiDividendDefaultURL
+ }
+ return baseURL + ssiDividendPath
+}
+
+// FetchDividendEvents returns only validated cash dividends and explicitly
+// described share dividends. SSI is queried with a one-calendar-day overlap in
+// Asia/Saigon, then results are filtered by publication time to (after, through].
+func (p *SSIDividendProvider) FetchDividendEvents(ctx context.Context, symbol string, after, through time.Time) ([]DividendEvent, error) {
+ symbol = strings.ToUpper(strings.TrimSpace(symbol))
+ if symbol == "" {
+ return nil, errors.New("stock: dividend symbol is empty")
+ }
+ if after.IsZero() || through.IsZero() || !through.After(after) {
+ return nil, errors.New("stock: invalid dividend event range")
+ }
+
+ from := after.In(saigonLocation).AddDate(0, 0, -1)
+ to := through.In(saigonLocation)
+ lowerBound := startOfSaigonDay(after).AddDate(0, 0, -1)
+ seen := make(map[string]struct{})
+ events := make([]DividendEvent, 0)
+ totalPages := 1
+ for page := 1; page <= totalPages; page++ {
+ if page > ssiDividendMaxPages {
+ return nil, fmt.Errorf("stock: SSI dividend response exceeds %d pages", ssiDividendMaxPages)
+ }
+ body, err := p.fetchPage(ctx, symbol, from, to, page)
+ if err != nil {
+ return nil, fmt.Errorf("stock: fetch SSI dividend page %d: %w", page, err)
+ }
+ if body.Paging.TotalPage > ssiDividendMaxPages {
+ return nil, fmt.Errorf("stock: SSI dividend response has %d pages, maximum is %d", body.Paging.TotalPage, ssiDividendMaxPages)
+ }
+ if body.Paging.TotalPage > totalPages {
+ totalPages = body.Paging.TotalPage
+ }
+ for _, raw := range body.Data {
+ id := strings.TrimSpace(raw.CorID)
+ if !ssiProviderIDPattern.MatchString(id) {
+ continue
+ }
+ if _, ok := seen[id]; ok {
+ continue
+ }
+ seen[id] = struct{}{}
+ event, ok := p.normalizeEvent(raw, symbol)
+ if !ok || event.PublishedAt.Before(lowerBound) || event.PublishedAt.After(through) {
+ continue
+ }
+ events = append(events, event)
+ }
+ }
+ sort.Slice(events, func(i, j int) bool {
+ if events[i].PublishedAt.Equal(events[j].PublishedAt) {
+ return events[i].ProviderID < events[j].ProviderID
+ }
+ return events[i].PublishedAt.Before(events[j].PublishedAt)
+ })
+ return events, nil
+}
+
+func startOfSaigonDay(value time.Time) time.Time {
+ local := value.In(saigonLocation)
+ return time.Date(local.Year(), local.Month(), local.Day(), 0, 0, 0, 0, saigonLocation)
+}
+
+func (p *SSIDividendProvider) fetchPage(ctx context.Context, symbol string, from, to time.Time, page int) (ssiDividendResponse, error) {
+ query := url.Values{}
+ query.Set("symbol", symbol)
+ query.Set("fromDate", from.Format("02/01/2006"))
+ query.Set("toDate", to.Format("02/01/2006"))
+ query.Set("page", strconv.Itoa(page))
+ query.Set("pageSize", strconv.Itoa(ssiDividendPageSize))
+ req, err := newSSIRequest(ctx, http.MethodGet, p.endpoint()+"?"+query.Encode(), nil)
+ if err != nil {
+ return ssiDividendResponse{}, err
+ }
+ resp, err := p.httpClient().Do(req)
+ if err != nil {
+ return ssiDividendResponse{}, err
+ }
+ defer func() { _ = resp.Body.Close() }()
+ if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+ snippet, _ := io.ReadAll(io.LimitReader(resp.Body, ssiErrorBodyLimit))
+ return ssiDividendResponse{}, fmt.Errorf("SSI status %d body %q", resp.StatusCode, strings.TrimSpace(string(snippet)))
+ }
+ var body ssiDividendResponse
+ if err := json.NewDecoder(resp.Body).Decode(&body); err != nil {
+ return ssiDividendResponse{}, fmt.Errorf("decode SSI response: %w", err)
+ }
+ if !strings.EqualFold(body.Code, "SUCCESS") || !strings.EqualFold(body.Status, "ok") {
+ return ssiDividendResponse{}, fmt.Errorf("SSI unsuccessful response code=%q status=%q", body.Code, body.Status)
+ }
+ if body.Paging.Page != 0 && body.Paging.Page != page {
+ return ssiDividendResponse{}, fmt.Errorf("SSI returned page %d for request page %d", body.Paging.Page, page)
+ }
+ return body, nil
+}
+
+func (p *SSIDividendProvider) normalizeEvent(raw ssiDividendEvent, requestedSymbol string) (DividendEvent, bool) {
+ symbol := strings.ToUpper(strings.TrimSpace(raw.Symbol))
+ if symbol != requestedSymbol {
+ return DividendEvent{}, false
+ }
+ publishedAt, exDate, recordDate, paymentDate, ok := parseSSIDividendDates(raw)
+ if !ok {
+ return DividendEvent{}, false
+ }
+ event := DividendEvent{
+ ProviderID: strings.TrimSpace(raw.CorID), Symbol: symbol,
+ PublishedAt: publishedAt, ExDate: exDate, RecordDate: recordDate, PaymentDate: paymentDate,
+ Title: strings.TrimSpace(raw.EventTitle), SourceURL: p.eventSourceURL(symbol, raw.CorID),
+ }
+ switch strings.ToUpper(strings.TrimSpace(raw.EventListCode)) {
+ case "DIV":
+ value, ok := positiveWholeNumber(string(raw.Value))
+ if !ok {
+ return DividendEvent{}, false
+ }
+ event.Kind = DividendKindCash
+ event.VNDPerShare = value
+ case "ISS":
+ wording := strings.ToLower(strings.Join([]string{raw.EventName, raw.EventTitle, raw.EventDescription}, " "))
+ if !strings.Contains(wording, "cổ tức") && !strings.Contains(wording, "dividend") {
+ return DividendEvent{}, false
+ }
+ owned, newShares, ok := exactShareRatio(string(raw.Ratio))
+ if !ok {
+ return DividendEvent{}, false
+ }
+ event.Kind = DividendKindShares
+ event.OwnedShares, event.NewShares = owned, newShares
+ default:
+ return DividendEvent{}, false
+ }
+ return event, true
+}
+
+// SSI normally supplies publicDate, the correct cursor timestamp. Older rows
+// can omit it, so ex-right, record, then issue/payment date are used as a
+// deterministic fallback. Any supplied but malformed date invalidates the row.
+func parseSSIDividendDates(raw ssiDividendEvent) (publishedAt, exDate, recordDate, paymentDate time.Time, ok bool) {
+ var valid bool
+ if exDate, valid = parseSSIOptionalDate(raw.ExrightDate); !valid {
+ return time.Time{}, time.Time{}, time.Time{}, time.Time{}, false
+ }
+ if recordDate, valid = parseSSIOptionalDate(raw.RecordDate); !valid {
+ return time.Time{}, time.Time{}, time.Time{}, time.Time{}, false
+ }
+ if paymentDate, valid = parseSSIOptionalDate(raw.IssueDate); !valid {
+ return time.Time{}, time.Time{}, time.Time{}, time.Time{}, false
+ }
+ if strings.TrimSpace(raw.PublicDate) != "" {
+ publishedAt, valid = parseSSIDate(raw.PublicDate)
+ if !valid {
+ return time.Time{}, time.Time{}, time.Time{}, time.Time{}, false
+ }
+ } else {
+ for _, fallback := range []time.Time{exDate, recordDate, paymentDate} {
+ if !fallback.IsZero() {
+ publishedAt = fallback
+ break
+ }
+ }
+ }
+ return publishedAt, exDate, recordDate, paymentDate, !publishedAt.IsZero()
+}
+
+func parseSSIOptionalDate(value string) (time.Time, bool) {
+ if strings.TrimSpace(value) == "" {
+ return time.Time{}, true
+ }
+ return parseSSIDate(value)
+}
+
+func parseSSIDate(value string) (time.Time, bool) {
+ value = strings.TrimSpace(value)
+ for _, layout := range []string{"02/01/2006 15:04:05", "02/01/2006", "2006-01-02T15:04:05"} {
+ parsed, err := time.ParseInLocation(layout, value, saigonLocation)
+ if err == nil {
+ return parsed, true
+ }
+ }
+ parsed, err := time.Parse(time.RFC3339, value)
+ return parsed, err == nil
+}
+
+func positiveWholeNumber(value string) (int64, bool) {
+ ratio, ok := new(big.Rat).SetString(strings.TrimSpace(value))
+ if !ok || ratio.Sign() <= 0 || !ratio.IsInt() || !ratio.Num().IsInt64() {
+ return 0, false
+ }
+ return ratio.Num().Int64(), true
+}
+
+func exactShareRatio(value string) (owned, newShares int64, ok bool) {
+ ratio, parsed := new(big.Rat).SetString(strings.TrimSpace(value))
+ if !parsed || ratio.Sign() <= 0 || !ratio.Num().IsInt64() || !ratio.Denom().IsInt64() {
+ return 0, 0, false
+ }
+ return ratio.Denom().Int64(), ratio.Num().Int64(), true
+}
+
+func (p *SSIDividendProvider) eventSourceURL(symbol, providerID string) string {
+ query := url.Values{}
+ query.Set("symbol", symbol)
+ query.Set("corId", strings.TrimSpace(providerID))
+ return p.endpoint() + "?" + query.Encode()
+}
diff --git a/internal/modules/stock/dividend_events_ssi_test.go b/internal/modules/stock/dividend_events_ssi_test.go
new file mode 100644
index 0000000..be5fc04
--- /dev/null
+++ b/internal/modules/stock/dividend_events_ssi_test.go
@@ -0,0 +1,157 @@
+package stock
+
+import (
+ "context"
+ "fmt"
+ "net/http"
+ "net/http/httptest"
+ "net/url"
+ "strconv"
+ "sync"
+ "testing"
+ "time"
+)
+
+func TestSSIDividendProviderPaginatesDeduplicatesFiltersAndNormalizes(t *testing.T) {
+ t.Parallel()
+ var mu sync.Mutex
+ requests := make([]url.Values, 0, 2)
+ server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ if r.URL.Path != ssiDividendPath {
+ t.Fatalf("path = %q, want %q", r.URL.Path, ssiDividendPath)
+ }
+ mu.Lock()
+ requests = append(requests, r.URL.Query())
+ mu.Unlock()
+ page, _ := strconv.Atoi(r.URL.Query().Get("page"))
+ w.Header().Set("Content-Type", "application/json")
+ switch page {
+ case 1:
+ _, _ = fmt.Fprint(w, `{"code":"SUCCESS","status":"ok","paging":{"totalPage":2,"page":1},"data":[
+ {"CorId":"cash","symbol":"TCB","eventListCode":"DIV","eventTitle":"Cash dividend","publicDate":"11/06/2026 00:00:00","exrightDate":"15/06/2026","recordDate":"16/06/2026","issueDate":"30/06/2026","value":"1500.0","ratio":"0"},
+ {"CorId":"not-dividend","symbol":"TCB","eventListCode":"ISS","eventTitle":"Rights offering","publicDate":"11/06/2026 00:00:00","ratio":"0.1"},
+ {"CorId":"wrong-symbol","symbol":"ACB","eventListCode":"DIV","eventTitle":"Cash dividend","publicDate":"11/06/2026 00:00:00","value":"500"}
+ ]}`)
+ case 2:
+ _, _ = fmt.Fprint(w, `{"code":"SUCCESS","status":"ok","paging":{"totalPage":2,"page":2},"data":[
+ {"CorId":"cash","symbol":"TCB","eventListCode":"DIV","eventTitle":"duplicate","publicDate":"11/06/2026 00:00:00","value":"999"},
+ {"CorId":"shares","symbol":"TCB","eventListCode":"ISS","eventName":"Trả cổ tức bằng cổ phiếu","eventTitle":"Share dividend","publicDate":"12/06/2026 12:00:00","exrightDate":"20/06/2026","ratio":0.125},
+ {"CorId":"invalid","symbol":"TCB","eventListCode":"DIV","eventTitle":"bad value","publicDate":"11/06/2026 00:00:00","value":"1.5"}
+ ]}`)
+ default:
+ t.Fatalf("unexpected page %d", page)
+ }
+ }))
+ defer server.Close()
+
+ provider := &SSIDividendProvider{HTTP: server.Client(), BaseURL: server.URL}
+ after := saigonTime(t, "10/06/2026 12:00:00")
+ through := saigonTime(t, "12/06/2026 12:00:00")
+ events, err := provider.FetchDividendEvents(context.Background(), " tcb ", after, through)
+ if err != nil {
+ t.Fatalf("FetchDividendEvents: %v", err)
+ }
+ if len(events) != 2 {
+ t.Fatalf("events = %+v, want cash and shares", events)
+ }
+ if cash := events[0]; cash.ProviderID != "cash" || cash.Kind != DividendKindCash || cash.VNDPerShare != 1500 || cash.Symbol != "TCB" {
+ t.Fatalf("cash = %+v", cash)
+ }
+ if shares := events[1]; shares.ProviderID != "shares" || shares.Kind != DividendKindShares || shares.OwnedShares != 8 || shares.NewShares != 1 {
+ t.Fatalf("shares = %+v", shares)
+ }
+ if events[1].PublishedAt != through {
+ t.Fatalf("through boundary event time = %v, want %v", events[1].PublishedAt, through)
+ }
+ if events[0].SourceURL == "" || events[0].Title != "Cash dividend" {
+ t.Fatalf("display metadata missing: %+v", events[0])
+ }
+
+ mu.Lock()
+ defer mu.Unlock()
+ if len(requests) != 2 {
+ t.Fatalf("requests = %d, want 2", len(requests))
+ }
+ for i, query := range requests {
+ if query.Get("symbol") != "TCB" || query.Get("fromDate") != "09/06/2026" || query.Get("toDate") != "12/06/2026" || query.Get("pageSize") != "50" || query.Get("page") != strconv.Itoa(i+1) {
+ t.Errorf("request %d query = %v", i+1, query)
+ }
+ }
+}
+
+func TestSSIDividendProviderUsesDayOverlapAndDateFallback(t *testing.T) {
+ t.Parallel()
+ server := dividendTestServer(t, http.StatusOK, `{"code":"SUCCESS","status":"ok","paging":{"totalPage":1,"page":1},"data":[
+ {"CorId":"too-old","symbol":"TCB","eventListCode":"DIV","publicDate":"08/06/2026 00:00:00","value":"100"},
+ {"CorId":"same-day-midnight","symbol":"TCB","eventListCode":"DIV","publicDate":"10/06/2026 00:00:00","value":"100"},
+ {"CorId":"at-after","symbol":"TCB","eventListCode":"DIV","publicDate":"10/06/2026 12:00:00","value":"100"},
+ {"CorId":"fallback","symbol":"TCB","eventListCode":"ISS","eventTitle":"Dividend in shares","exrightDate":"11/06/2026","ratio":"0.13"},
+ {"CorId":"after-through","symbol":"TCB","eventListCode":"DIV","publicDate":"13/06/2026 00:00:00","value":"100"},
+ {"CorId":"bad-date","symbol":"TCB","eventListCode":"DIV","publicDate":"not-a-date","value":"100"}
+ ]}`)
+ defer server.Close()
+ provider := &SSIDividendProvider{HTTP: server.Client(), BaseURL: server.URL}
+ events, err := provider.FetchDividendEvents(context.Background(), "TCB", saigonTime(t, "10/06/2026 12:00:00"), saigonTime(t, "12/06/2026 00:00:00"))
+ if err != nil {
+ t.Fatalf("FetchDividendEvents: %v", err)
+ }
+ if len(events) != 3 || events[0].ProviderID != "same-day-midnight" || events[1].ProviderID != "at-after" || events[2].ProviderID != "fallback" || events[2].PublishedAt != saigonTime(t, "11/06/2026 00:00:00") {
+ t.Fatalf("events = %+v, want same-day overlap, cursor boundary, and fallback events", events)
+ }
+ if events[2].OwnedShares != 100 || events[2].NewShares != 13 {
+ t.Fatalf("ratio = %d:%d, want 100:13", events[2].OwnedShares, events[2].NewShares)
+ }
+}
+
+func TestSSIDividendProviderRequiresEveryPage(t *testing.T) {
+ t.Parallel()
+ server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ if r.URL.Query().Get("page") == "2" {
+ http.Error(w, "upstream unavailable", http.StatusServiceUnavailable)
+ return
+ }
+ _, _ = fmt.Fprint(w, `{"code":"SUCCESS","status":"ok","paging":{"totalPage":2,"page":1},"data":[{"CorId":"cash","symbol":"TCB","eventListCode":"DIV","publicDate":"11/06/2026","value":"100"}]}`)
+ }))
+ defer server.Close()
+ provider := &SSIDividendProvider{HTTP: server.Client(), BaseURL: server.URL}
+ events, err := provider.FetchDividendEvents(context.Background(), "TCB", saigonTime(t, "10/06/2026"), saigonTime(t, "12/06/2026"))
+ if err == nil || events != nil {
+ t.Fatalf("events, err = %+v, %v; want nil and page error", events, err)
+ }
+}
+
+func TestSSIDividendProviderRejectsInvalidRangeAndExcessivePages(t *testing.T) {
+ t.Parallel()
+ provider := &SSIDividendProvider{}
+ now := time.Now()
+ if _, err := provider.FetchDividendEvents(context.Background(), "TCB", now, now); err == nil {
+ t.Fatal("equal cursor range accepted")
+ }
+
+ server := dividendTestServer(t, http.StatusOK, `{"code":"SUCCESS","status":"ok","paging":{"totalPage":21,"page":1},"data":[]}`)
+ defer server.Close()
+ provider = &SSIDividendProvider{HTTP: server.Client(), BaseURL: server.URL}
+ if _, err := provider.FetchDividendEvents(context.Background(), "TCB", saigonTime(t, "10/06/2026"), saigonTime(t, "12/06/2026")); err == nil {
+ t.Fatal("response above page cap accepted")
+ }
+}
+
+func dividendTestServer(t *testing.T, status int, body string) *httptest.Server {
+ t.Helper()
+ return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+ w.WriteHeader(status)
+ _, _ = fmt.Fprint(w, body)
+ }))
+}
+
+func saigonTime(t *testing.T, value string) time.Time {
+ t.Helper()
+ for _, layout := range []string{"02/01/2006 15:04:05", "02/01/2006"} {
+ parsed, err := time.ParseInLocation(layout, value, saigonLocation)
+ if err == nil {
+ return parsed
+ }
+ }
+ t.Fatalf("parse test time %q", value)
+ return time.Time{}
+}
diff --git a/internal/modules/stock/dividend_flow_test.go b/internal/modules/stock/dividend_flow_test.go
new file mode 100644
index 0000000..29dba3a
--- /dev/null
+++ b/internal/modules/stock/dividend_flow_test.go
@@ -0,0 +1,379 @@
+package stock
+
+import (
+ "context"
+ "errors"
+ "net/http"
+ "net/http/httptest"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/go-telegram/bot/models"
+
+ "github.com/tiennm99/miti99bot/internal/storage"
+ "github.com/tiennm99/miti99bot/internal/testutil"
+)
+
+type fakeDividendProvider struct {
+ events []DividendEvent
+ err error
+ calls int
+}
+
+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++
+ return append([]DividendEvent(nil), f.events...), f.err
+}
+
+func newDividendFlowState(t *testing.T, events []DividendEvent) (*state, Store, PendingDividendStore, *testutil.RecordingBot, time.Time) {
+ t.Helper()
+ now := time.Date(2026, 6, 25, 12, 0, 0, 0, saigonLocation)
+ priceServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+ w.Header().Set("Content-Type", "application/json")
+ _, _ = w.Write([]byte(`{"data":[{"stockSymbol":"TCB","matchedPrice":30000}]}`))
+ }))
+ t.Cleanup(priceServer.Close)
+
+ provider := storage.NewMemoryProvider()
+ collection := provider.Collection(CollectionName)
+ store := storage.Typed[Portfolio](collection)
+ pending := storage.Typed[PendingDividendAction](collection)
+ s := &state{
+ store: store,
+ pending: pending,
+ prices: &PriceClient{HTTP: priceServer.Client(), URL: priceServer.URL},
+ dividends: &fakeDividendProvider{events: events},
+ nowFn: func() time.Time { return now },
+ newDividendToken: func() (string, error) {
+ return "abcdefghijklmnopqrstuv", nil
+ },
+ }
+ rb := testutil.NewRecordingBot(t)
+ return s, store, pending, rb, now
+}
+
+func seedDividendFlowPortfolio(t *testing.T, store Store, now time.Time, quantity int64) {
+ t.Helper()
+ p := NewPortfolio(now.Add(-48 * time.Hour).UnixMilli())
+ p.VND = 100_000
+ p.Meta.Invested = 3_000_000
+ if err := p.BuyTicker("TCB", quantity, float64(quantity)*30_000, now.Add(-48*time.Hour).UnixMilli()); err != nil {
+ t.Fatal(err)
+ }
+ if err := SavePortfolio(context.Background(), store, 7, p); err != nil {
+ t.Fatal(err)
+ }
+}
+
+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{
+ 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),
+ VNDPerShare: 1500, Title: "Cash dividend",
+ }
+ s, store, pending, rb, now := 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)
+ }
+ 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)
+ }
+}
+
+func TestDividendCallback_OnlyOwnerCanApplyAndUsesCurrentHolding(t *testing.T) {
+ s, store, pending, rb, now := newDividendFlowState(t, nil)
+ 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)
+ }
+ 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 {
+ 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)
+ }
+ if _, _, err := pending.Get(context.Background(), key); !errors.Is(err, storage.ErrNotFound) {
+ t.Fatalf("stale action was not invalidated: %v", err)
+ }
+}
+
+func TestDividendCallback_ExpiredActionDoesNotApply(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.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)); 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)
+ }
+}
+
+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 {
+ t.Fatal(err)
+ }
+ 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 {
+ 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"])
+ }
+}
+
+func TestConcurrentDividendSuggestionsCreateOneButton(t *testing.T) {
+ s, store, pending, rb, now := newDividendFlowState(t, nil)
+ seedDividendFlowPortfolio(t, store, now, 100)
+ event := DividendEvent{ProviderID: "2612974", Symbol: "TCB", Kind: DividendKindCash, VNDPerShare: 1500}
+ msg := testutil.NewPrivateMessage(7, "/stock_portfolio").Message
+ 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)
+ }()
+ }
+ wg.Wait()
+ close(errs)
+ for err := range errs {
+ if err != nil {
+ 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)
+ }
+}
+
+func TestDividendFetchDoesNotBindOldEventToReopenedPosition(t *testing.T) {
+ s, store, pending, rb, now := 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)
+ }
+ if err := SavePortfolio(context.Background(), store, 7, p); err != nil {
+ t.Fatal(err)
+ }
+ close(blocking.release)
+ if err := <-done; 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)
+ }
+ 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)
+ }
+}
+
+func dividendCallbackUpdate(userID, chatID int64, messageID int) *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: messageID, Chat: models.Chat{ID: chatID, Type: models.ChatTypePrivate},
+ }},
+ }}
+}
diff --git a/internal/modules/stock/dividend_notifications.go b/internal/modules/stock/dividend_notifications.go
new file mode 100644
index 0000000..d416d1e
--- /dev/null
+++ b/internal/modules/stock/dividend_notifications.go
@@ -0,0 +1,285 @@
+package stock
+
+import (
+ "context"
+ "fmt"
+ "html"
+ "sort"
+ "strconv"
+ "strings"
+ "sync"
+ "time"
+ "unicode/utf8"
+
+ "github.com/go-telegram/bot"
+ "github.com/go-telegram/bot/models"
+
+ "github.com/tiennm99/miti99bot/internal/log"
+ "github.com/tiennm99/miti99bot/internal/modules/util/chathelper"
+)
+
+const (
+ dividendFetchTimeout = 12 * time.Second
+ dividendFetchWorkers = 4
+)
+
+type dividendCheckResult struct {
+ symbol string
+ openedAt int64
+ events []DividendEvent
+ err error
+}
+
+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 holding struct {
+ symbol string
+ qty int64
+ after time.Time
+ }
+ 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 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, events: events, err: err}
+ }()
+ }
+ wg.Wait()
+
+ successful := make([]string, 0, len(results))
+ failed := make([]string, 0)
+ for _, result := range results {
+ if result.err != nil {
+ log.Error("stock_fetch_dividend_events", "ticker", result.symbol, "err", result.err)
+ failed = append(failed, result.symbol)
+ continue
+ }
+ 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
+ }
+ }
+ if delivered {
+ successful = append(successful, result.symbol)
+ } else {
+ failed = append(failed, result.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)
+ return chathelper.Reply(ctx, b, msg, "Dividend events could not be checked for "+strings.Join(failed, ", ")+". Try again later.")
+ }
+ return nil
+}
+
+func (s *state) advanceDividendCursors(ctx context.Context, userID int64, symbols []string, checkedThrough int64) error {
+ defer s.locks.Acquire(strconv.FormatInt(userID, 10))()
+ p, err := LoadPortfolio(ctx, s.store, userID, checkedThrough)
+ if err != nil {
+ return 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
+ }
+ }
+ if !changed {
+ return nil
+ }
+ return SavePortfolio(ctx, s.store, userID, p)
+}
+
+func (s *state) sendDividendSuggestion(ctx context.Context, b *bot.Bot, msg *models.Message, userID, expectedOpenedAt int64, checkedThrough time.Time, event DividendEvent) error {
+ defer s.locks.Acquire(strconv.FormatInt(userID, 10))()
+
+ now := s.now()
+ latest, err := LoadPortfolio(ctx, s.store, userID, now.UnixMilli())
+ if err != nil {
+ return err
+ }
+ if _, applied := latest.AppliedDividendEvents[dividendLedgerKey(event.ProviderID)]; applied {
+ return nil
+ }
+ position, held := latest.Assets[event.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 != nil {
+ return err
+ }
+ if hasPending {
+ 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(),
+ }
+ token, err := s.createPendingDividend(ctx, action)
+ if err != nil {
+ return err
+ }
+ key := pendingDividendKey(token)
+ sent, err := b.SendMessage(ctx, &bot.SendMessageParams{
+ ChatID: msg.Chat.ID,
+ MessageThreadID: msg.MessageThreadID,
+ Text: dividendEventText(event, observedHolding),
+ ParseMode: models.ParseModeHTML,
+ ReplyMarkup: &models.InlineKeyboardMarkup{InlineKeyboard: [][]models.InlineKeyboardButton{{{
+ Text: "Apply dividend",
+ CallbackData: dividendCallbackPrefix + token,
+ }}}},
+ })
+ if err != nil {
+ _ = s.pending.Delete(ctx, key)
+ return err
+ }
+ action.MessageID = sent.ID
+ if err := s.pending.Put(ctx, key, action); err != nil {
+ _ = removeDividendButton(ctx, b, action.ChatID, action.MessageID)
+ _ = s.pending.Delete(ctx, key)
+ return fmt.Errorf("bind dividend action message: %w", err)
+ }
+ 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 {
+ var value string
+ switch event.Kind {
+ case DividendKindCash:
+ value = FormatVND(float64(event.VNDPerShare)) + "/share"
+ case DividendKindShares:
+ value = strconv.FormatInt(event.OwnedShares, 10) + ":" + strconv.FormatInt(event.NewShares, 10) + " shares"
+ }
+ lines := []string{
+ "Recent dividend event · " + html.EscapeString(event.Symbol) + "",
+ 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"))
+ }
+ if !event.PaymentDate.IsZero() {
+ lines = append(lines, "Payment/trading: "+event.PaymentDate.Format("02/01/2006"))
+ }
+ if title := truncateRunes(strings.TrimSpace(event.Title), 240); title != "" {
+ lines = append(lines, html.EscapeString(title))
+ }
+ lines = append(lines,
+ "SSI event: "+html.EscapeString(event.ProviderID)+"",
+ "Observed holding: "+html.EscapeString(formatShareQuantity(observedHolding))+" shares.",
+ "The dividend uses your current holding when you accept.",
+ "Button expires in 24 hours.",
+ )
+ return strings.Join(lines, "\n")
+}
+
+func truncateRunes(value string, limit int) string {
+ if utf8.RuneCountInString(value) <= limit {
+ return value
+ }
+ runes := []rune(value)
+ return string(runes[:limit-1]) + "…"
+}
+
+func uniqueStrings(values []string) []string {
+ if len(values) < 2 {
+ return values
+ }
+ out := values[:1]
+ for _, value := range values[1:] {
+ if value != out[len(out)-1] {
+ out = append(out, value)
+ }
+ }
+ return out
+}
diff --git a/internal/modules/stock/handlers.go b/internal/modules/stock/handlers.go
index 41789d5..cd53d2e 100644
--- a/internal/modules/stock/handlers.go
+++ b/internal/modules/stock/handlers.go
@@ -21,10 +21,13 @@ import (
// prefixes/partitions). PriceClient is reused across calls; nowFn allows
// tests to inject a deterministic clock for portfolio CreatedAt.
type state struct {
- store Store
- prices *PriceClient
- locks keylock.Map
- nowFn func() time.Time
+ store Store
+ pending PendingDividendStore
+ prices *PriceClient
+ dividends DividendEventProvider
+ locks keylock.Map
+ nowFn func() time.Time
+ newDividendToken func() (string, error)
}
func (s *state) now() time.Time {
@@ -35,10 +38,12 @@ func (s *state) now() time.Time {
}
// newState builds the default state used by the module factory.
-func newState(store Store) *state {
+func newState(store Store, pending PendingDividendStore) *state {
return &state{
- store: store,
- prices: &PriceClient{},
+ store: store,
+ pending: pending,
+ prices: &PriceClient{},
+ dividends: &SSIDividendProvider{},
}
}
@@ -452,15 +457,17 @@ func (s *state) handleDividend(ctx context.Context, b *bot.Bot, update *models.U
"\nBalance: "+FormatVND(balance))
}
-// handleStats fetches current prices for held tickers and renders the
-// portfolio. Read-only; no portfolio mutation, so no keylock.
+// 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.
func (s *state) handleStats(ctx context.Context, b *bot.Bot, update *models.Update) error {
userID, ok := senderInfo(update)
if !ok {
return chathelper.Reply(ctx, b, update.Message,
"Cannot identify user — /stock_portfolio needs a sender.")
}
- p, err := LoadPortfolio(ctx, s.store, userID, s.now().UnixMilli())
+ checkedThrough := s.now()
+ p, err := LoadPortfolio(ctx, s.store, userID, checkedThrough.UnixMilli())
if err != nil {
log.Error("stock_load_portfolio", "user", userID, "err", err)
return chathelper.Reply(ctx, b, update.Message, "Could not load portfolio. Try again later.")
@@ -486,13 +493,13 @@ func (s *state) handleStats(ctx context.Context, b *bot.Bot, update *models.Upda
if len(heldList) > 0 {
fetchCtx, cancel := chathelper.FetchContext(ctx)
- defer cancel()
symbols := make([]string, 0, len(heldList))
for _, h := range heldList {
symbols = append(symbols, h.symbol)
}
prices, fetchErr := s.prices.FetchPrices(fetchCtx, symbols)
+ cancel()
if fetchErr != nil {
log.Error("stock_fetch_prices", "symbols", strings.Join(symbols, ","), "err", fetchErr)
}
@@ -535,7 +542,10 @@ func (s *state) handleStats(ctx context.Context, b *bot.Bot, update *models.Upda
{"Account P&L", FormatPnL(totalValue, p.Meta.Invested)},
}
}
- return chathelper.ReplyHTML(ctx, b, update.Message, portfolioTableReply("Stock Portfolio", positions, summary))
+ if err := chathelper.ReplyHTML(ctx, b, update.Message, portfolioTableReply("Stock Portfolio", positions, summary)); err != nil {
+ return err
+ }
+ return s.notifyDividendEvents(ctx, b, update.Message, userID, p, checkedThrough)
}
const portfolioReplyLimit = 4000
diff --git a/internal/modules/stock/pending_dividend.go b/internal/modules/stock/pending_dividend.go
new file mode 100644
index 0000000..2ec4018
--- /dev/null
+++ b/internal/modules/stock/pending_dividend.go
@@ -0,0 +1,112 @@
+package stock
+
+import (
+ "context"
+ "crypto/rand"
+ "crypto/sha256"
+ "encoding/base64"
+ "errors"
+ "fmt"
+ "regexp"
+ "time"
+
+ "github.com/tiennm99/miti99bot/internal/storage"
+)
+
+const (
+ dividendCallbackPrefix = "stock_div:"
+ pendingDividendPrefix = "pending-dividend:"
+ pendingDividendTTL = 24 * time.Hour
+ dividendTokenBytes = 16
+)
+
+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.
+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"`
+}
+
+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 {
+ return "", fmt.Errorf("generate dividend action token: %w", err)
+ }
+ return base64.RawURLEncoding.EncodeToString(raw), nil
+}
+
+func callbackToken(data string) (string, bool) {
+ if len(data) != len(dividendCallbackPrefix)+22 || data[:len(dividendCallbackPrefix)] != dividendCallbackPrefix {
+ return "", false
+ }
+ token := data[len(dividendCallbackPrefix):]
+ return token, dividendTokenPattern.MatchString(token)
+}
+
+func (s *state) createPendingDividend(ctx context.Context, action PendingDividendAction) (string, error) {
+ if s.pending == nil {
+ return "", errors.New("stock: pending dividend store unavailable")
+ }
+ generator := s.newDividendToken
+ if generator == nil {
+ generator = generateDividendToken
+ }
+ for attempt := 0; attempt < 3; attempt++ {
+ token, err := generator()
+ if err != nil {
+ return "", err
+ }
+ if !dividendTokenPattern.MatchString(token) {
+ return "", errors.New("stock: invalid generated dividend token")
+ }
+ err = s.pending.PutVersioned(ctx, pendingDividendKey(token), 0, action)
+ if errors.Is(err, storage.ErrConflict) {
+ continue
+ }
+ if err != nil {
+ return "", fmt.Errorf("save pending dividend action: %w", err)
+ }
+ return token, nil
+ }
+ return "", errors.New("stock: could not allocate unique dividend token")
+}
+
+func (s *state) cleanupExpiredDividends(ctx context.Context, now int64) {
+ if s.pending == nil {
+ return
+ }
+ keys, err := s.pending.List(ctx, pendingDividendPrefix)
+ if err != nil {
+ return
+ }
+ for _, key := range keys {
+ action, _, getErr := s.pending.Get(ctx, key)
+ if getErr == nil && action.ExpiresAt > 0 && action.ExpiresAt <= now {
+ _ = s.pending.Delete(ctx, key)
+ }
+ }
+}
diff --git a/internal/modules/stock/portfolio.go b/internal/modules/stock/portfolio.go
index 1c16399..7aacb62 100644
--- a/internal/modules/stock/portfolio.go
+++ b/internal/modules/stock/portfolio.go
@@ -21,12 +21,14 @@ 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"`
}
type Portfolio struct {
- VND float64 `json:"vnd" bson:"vnd"`
- Assets map[string]AssetPosition `json:"assets" bson:"assets"`
- Meta PortfolioMeta `json:"meta" bson:"meta"`
+ 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"`
}
type PortfolioMeta struct {
@@ -49,6 +51,9 @@ 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.Meta.CreatedAt == 0 {
p.Meta.CreatedAt = now
}
@@ -82,10 +87,15 @@ 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 {
+ if position.Quantity <= 0 || !isPositiveFiniteCost(position.Base) || position.DividendCheckedAt <= 0 || 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")
+ }
+ }
return nil
}
@@ -111,6 +121,7 @@ func (p *Portfolio) BuyTicker(symbol string, quantity int64, base float64, now i
p.Assets = map[string]AssetPosition{}
}
position := p.Assets[symbol]
+ isOpening := position.Quantity == 0
if position.Quantity > math.MaxInt64-quantity {
return fmt.Errorf("stock: quantity overflows")
}
@@ -122,6 +133,9 @@ func (p *Portfolio) BuyTicker(symbol string, quantity int64, base float64, now i
if position.DividendCheckedAt == 0 {
position.DividendCheckedAt = now
}
+ if isOpening {
+ position.OpenedAt = now
+ }
p.Assets[symbol] = position
return nil
}
diff --git a/internal/modules/stock/stock.go b/internal/modules/stock/stock.go
index 264b488..8a69737 100644
--- a/internal/modules/stock/stock.go
+++ b/internal/modules/stock/stock.go
@@ -7,8 +7,12 @@ import (
// New is the stock module Factory. Eight user-facing commands.
func New(deps modules.Deps) modules.Module {
- s := newState(storage.Typed[Portfolio](deps.Store))
+ s := newState(
+ storage.Typed[Portfolio](deps.Store),
+ storage.Typed[PendingDividendAction](deps.Store),
+ )
return modules.Module{
+ Callbacks: []modules.Callback{{Prefix: dividendCallbackPrefix, Visibility: modules.VisibilityPublic, Handler: s.handleDividendCallback}},
Commands: []modules.Command{
{
Name: "stock_price",
diff --git a/internal/modules/validate.go b/internal/modules/validate.go
index ac34444..ce6de82 100644
--- a/internal/modules/validate.go
+++ b/internal/modules/validate.go
@@ -47,3 +47,21 @@ func validateCron(c Cron) error {
}
return nil
}
+
+func validateCallback(c Callback) error {
+ if strings.TrimSpace(c.Prefix) == "" {
+ return fmt.Errorf("callback: prefix is required")
+ }
+ if len(c.Prefix) > 32 || strings.ContainsAny(c.Prefix, "\r\n\x00") {
+ return fmt.Errorf("callback prefix %q is invalid", c.Prefix)
+ }
+ switch c.Visibility {
+ case VisibilityPublic, VisibilityProtected, VisibilityPrivate:
+ default:
+ return fmt.Errorf("callback prefix %q: unknown visibility %d", c.Prefix, c.Visibility)
+ }
+ if c.Handler == nil {
+ return fmt.Errorf("callback prefix %q: handler is nil", c.Prefix)
+ }
+ return nil
+}