From 77e8ca05d6bd52254c5eae849cd5fbe67eeaf099 Mon Sep 17 00:00:00 2001 From: tiennm99 Date: Tue, 21 Jul 2026 18:19:50 +0700 Subject: [PATCH] feat(stock): add dividend event buttons --- internal/modules/dispatcher.go | 30 +- internal/modules/dispatcher_test.go | 4 + internal/modules/module.go | 13 + internal/modules/registry.go | 24 ++ internal/modules/registry_test.go | 24 ++ internal/modules/stock/dividend_callback.go | 196 +++++++++ internal/modules/stock/dividend_events.go | 42 ++ internal/modules/stock/dividend_events_ssi.go | 314 +++++++++++++++ .../modules/stock/dividend_events_ssi_test.go | 157 ++++++++ internal/modules/stock/dividend_flow_test.go | 379 ++++++++++++++++++ .../modules/stock/dividend_notifications.go | 285 +++++++++++++ internal/modules/stock/handlers.go | 34 +- internal/modules/stock/pending_dividend.go | 112 ++++++ internal/modules/stock/portfolio.go | 22 +- internal/modules/stock/stock.go | 6 +- internal/modules/validate.go | 18 + 16 files changed, 1641 insertions(+), 19 deletions(-) create mode 100644 internal/modules/stock/dividend_callback.go create mode 100644 internal/modules/stock/dividend_events.go create mode 100644 internal/modules/stock/dividend_events_ssi.go create mode 100644 internal/modules/stock/dividend_events_ssi_test.go create mode 100644 internal/modules/stock/dividend_flow_test.go create mode 100644 internal/modules/stock/dividend_notifications.go create mode 100644 internal/modules/stock/pending_dividend.go 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 +}