From 46eed875a94288615f48fa8696e23e0012051484 Mon Sep 17 00:00:00 2001 From: Flechazo <2558755403@qq.com> Date: Sat, 3 Oct 2026 00:56:48 +0800 Subject: [PATCH] feat(server): queue dynamic operator mail grants --- go/cmd/bd2server/main.go | 6 + go/internal/server/mail/grant_spool.go | 201 +++++++++++++ go/internal/server/mail/grant_spool_test.go | 299 ++++++++++++++++++++ go/internal/server/mail/mail.go | 116 +++++--- tools/python/dev_mail_grant.py | 154 +++++++++- tools/python/tests/test_dev_mail_grant.py | 178 ++++++++++++ 6 files changed, 901 insertions(+), 53 deletions(-) create mode 100644 go/internal/server/mail/grant_spool.go create mode 100644 go/internal/server/mail/grant_spool_test.go create mode 100644 tools/python/tests/test_dev_mail_grant.py diff --git a/go/cmd/bd2server/main.go b/go/cmd/bd2server/main.go index df2dffa..6c9493a 100644 --- a/go/cmd/bd2server/main.go +++ b/go/cmd/bd2server/main.go @@ -82,6 +82,7 @@ func serve(args []string) (serveErr error) { playerSeed := fs.String("player-seed", "", "versioned starter inventory and characters") readonlySeed := fs.String("readonly-seed", "", "versioned server schedules and optional feature defaults") mailSeed := fs.String("mail-seed", "", "versioned starter mailbox") + mailGrantSpool := fs.String("mail-grant-spool", "", "optional local JSON spool for idempotent dynamic system mail") stateFile := fs.String("state", "", "account SQLite database override") deckSeed := fs.String("deck-seed", "", "versioned starter deck") worldSeed := fs.String("world-seed", "", "versioned starter world") @@ -340,6 +341,11 @@ func serve(args []string) (serveErr error) { if err := mailService.AttachSeedPath(filepath.Clean(*mailSeed)); err != nil { return fmt.Errorf("watch mail seed: %w", err) } + if *mailGrantSpool != "" { + if err := mailService.AttachGrantSpoolPath(*mailGrantSpool); err != nil { + return fmt.Errorf("attach mail grant spool: %w", err) + } + } missionDesign, err := gamedata.LoadMissionDesign(gameData, *gameDataVersion) if err != nil { return fmt.Errorf("load mission GameData: %w", err) diff --git a/go/internal/server/mail/grant_spool.go b/go/internal/server/mail/grant_spool.go new file mode 100644 index 0000000..6ef0cf4 --- /dev/null +++ b/go/internal/server/mail/grant_spool.go @@ -0,0 +1,201 @@ +package mail + +import ( + "bytes" + "encoding/json" + "errors" + "fmt" + "io" + "math" + "os" + "path/filepath" + "strings" + "time" + "unicode/utf8" + + "bd2server/internal/server/gamedata" +) + +// GrantSpool is a local operator input, independent of the versioned starter +// mailbox and player state. Writers atomically replace the whole JSON file. +// Version 1 uses Unix milliseconds; a stable identity is never issued twice, +// including after the player opens the mail or the server restarts. +type GrantSpool struct { + Version uint64 `json:"version"` + Grants []Grant `json:"grants"` +} + +type Grant struct { + Identity string `json:"identity"` + Title string `json:"title"` + Body string `json:"body"` + SentAt int64 `json:"sent_at"` + Rewards []GrantReward `json:"rewards"` +} + +type GrantReward struct { + Type uint64 `json:"type"` + ID uint64 `json:"id"` + Count uint64 `json:"count"` +} + +type compensation struct { + identity, title, body string + sentAt time.Time + rewards []gamedata.Reward +} + +func (g compensation) validate() error { + if strings.TrimSpace(g.identity) == "" || strings.TrimSpace(g.identity) != g.identity || utf8.RuneCountInString(g.identity) > 500 || + strings.TrimSpace(g.title) == "" || utf8.RuneCountInString(g.title) > 500 || + strings.TrimSpace(g.body) == "" || utf8.RuneCountInString(g.body) > 5000 || + g.sentAt.IsZero() || g.sentAt.UnixMilli() <= 0 || g.sentAt.UnixMilli() > math.MaxInt64-int64(30*24*time.Hour/time.Millisecond) || len(g.rewards) == 0 { + return errors.New("mail: invalid compensation identity, content, time or rewards") + } + for _, reward := range g.rewards { + if reward.Count == 0 || reward.Count > math.MaxInt32 || (!currencyRewardTypes[reward.Type] && (!itemDBInfoTypes[reward.Type] || reward.ID == 0)) || + (currencyRewardTypes[reward.Type] && reward.ID != 0) { + return fmt.Errorf("mail: compensation %q has unsupported reward type=%d id=%d count=%d", g.identity, reward.Type, reward.ID, reward.Count) + } + } + return nil +} + +// AttachGrantSpoolPath enables reads before each /MailInfo. An absent file is +// an empty queue, allowing an operator to create it after server startup. +// Existing malformed input is rejected immediately without issuing mail. +func (s *Service) AttachGrantSpoolPath(path string) error { + if s == nil || strings.TrimSpace(path) == "" { + return errors.New("mail: invalid grant spool path") + } + abs, err := filepath.Abs(filepath.Clean(path)) + if err != nil { + return fmt.Errorf("mail: resolve grant spool path: %w", err) + } + if _, err := readGrantSpool(abs); err != nil { + return err + } + s.mu.Lock() + defer s.mu.Unlock() + s.grantSpoolPath = abs + return nil +} + +func readGrantSpool(path string) ([]compensation, error) { + if path == "" { + return nil, nil + } + data, err := os.ReadFile(path) + if os.IsNotExist(err) { + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("mail: read grant spool: %w", err) + } + grants, err := decodeGrantSpool(data) + if err != nil { + return nil, fmt.Errorf("mail: reject grant spool: %w", err) + } + return grants, nil +} + +// Decode the entire file before returning any grants. The exact object reader +// rejects missing, unknown, duplicate and null fields at every schema level. +func decodeGrantSpool(data []byte) ([]compensation, error) { + if !utf8.Valid(data) { + return nil, errors.New("invalid UTF-8") + } + root, err := exactSpoolObject(data, "version", "grants") + if err != nil { + return nil, err + } + var version uint64 + if err := json.Unmarshal(root["version"], &version); err != nil || version != 1 { + return nil, errors.New("unsupported grant spool version") + } + var rawGrants []json.RawMessage + if err := json.Unmarshal(root["grants"], &rawGrants); err != nil { + return nil, err + } + seen := make(map[string]bool, len(rawGrants)) + grants := make([]compensation, 0, len(rawGrants)) + for i, raw := range rawGrants { + fields, err := exactSpoolObject(raw, "identity", "title", "body", "sent_at", "rewards") + if err != nil { + return nil, fmt.Errorf("grant %d: %w", i, err) + } + var entry Grant + if err := json.Unmarshal(raw, &entry); err != nil { + return nil, fmt.Errorf("grant %d: %w", i, err) + } + if seen[entry.Identity] { + return nil, fmt.Errorf("duplicate grant identity %q", entry.Identity) + } + seen[entry.Identity] = true + var rawRewards []json.RawMessage + if err := json.Unmarshal(fields["rewards"], &rawRewards); err != nil { + return nil, err + } + grant := compensation{identity: entry.Identity, title: entry.Title, body: entry.Body, sentAt: time.UnixMilli(entry.SentAt)} + for j, reward := range rawRewards { + if _, err := exactSpoolObject(reward, "type", "id", "count"); err != nil { + return nil, fmt.Errorf("grant %d reward %d: %w", i, j, err) + } + grant.rewards = append(grant.rewards, gamedata.Reward{Type: entry.Rewards[j].Type, ID: entry.Rewards[j].ID, Count: entry.Rewards[j].Count}) + } + if err := grant.validate(); err != nil { + return nil, fmt.Errorf("grant %d: %w", i, err) + } + grants = append(grants, grant) + } + return grants, nil +} + +func exactSpoolObject(data []byte, names ...string) (map[string]json.RawMessage, error) { + decoder := json.NewDecoder(bytes.NewReader(data)) + token, err := decoder.Token() + if err != nil { + return nil, err + } + if token != json.Delim('{') { + return nil, errors.New("expected JSON object") + } + wanted := make(map[string]bool, len(names)) + for _, name := range names { + wanted[name] = true + } + fields := make(map[string]json.RawMessage, len(names)) + for decoder.More() { + token, err := decoder.Token() + if err != nil { + return nil, err + } + name, ok := token.(string) + if !ok || !wanted[name] { + return nil, fmt.Errorf("unknown field %q", token) + } + if _, exists := fields[name]; exists { + return nil, fmt.Errorf("duplicate field %q", name) + } + var value json.RawMessage + if err := decoder.Decode(&value); err != nil { + return nil, err + } + if bytes.Equal(bytes.TrimSpace(value), []byte("null")) { + return nil, fmt.Errorf("null field %q", name) + } + fields[name] = value + } + if _, err := decoder.Token(); err != nil { + return nil, err + } + if _, err := decoder.Token(); err != io.EOF { + return nil, errors.New("trailing JSON content") + } + for _, name := range names { + if _, exists := fields[name]; !exists { + return nil, fmt.Errorf("missing field %q", name) + } + } + return fields, nil +} diff --git a/go/internal/server/mail/grant_spool_test.go b/go/internal/server/mail/grant_spool_test.go new file mode 100644 index 0000000..90fa742 --- /dev/null +++ b/go/internal/server/mail/grant_spool_test.go @@ -0,0 +1,299 @@ +package mail + +import ( + "encoding/json" + "errors" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "bd2server/internal/server/accountstate" + "bd2server/internal/server/player" + "bd2server/internal/server/stateio" + "bd2server/internal/server/wire" +) + +func spoolTestService(t *testing.T, store stateio.Store) (*Service, *player.Inventory, *player.Wallet) { + t.Helper() + inventory, err := player.OpenInventory(store, &player.Starter{Version: "2.35.10"}) + if err != nil { + t.Fatal(err) + } + wallet, err := player.OpenWallet(store, player.Currency{}) + if err != nil { + t.Fatal(err) + } + service, err := OpenService(store, &Starter{Version: "2.35.10", MailCount: 1, MaxMailID: 100}, inventory, wallet) + if err != nil { + t.Fatal(err) + } + return service, inventory, wallet +} + +func spoolGrant(identity string, rewards ...GrantReward) Grant { + return Grant{Identity: identity, Title: "系统奖励", Body: "请领取邮件附件。", SentAt: time.Date(2026, 10, 3, 1, 0, 0, 0, time.UTC).UnixMilli(), Rewards: rewards} +} + +func writeSpool(t *testing.T, path string, grants ...Grant) { + t.Helper() + data, err := json.Marshal(GrantSpool{Version: 1, Grants: grants}) + if err != nil { + t.Fatal(err) + } + if err := os.WriteFile(path, data, 0o600); err != nil { + t.Fatal(err) + } +} + +func TestGrantSpoolImportsDynamicMailOnceAcrossOpenAndRestart(t *testing.T) { + store := stateio.NewMemory() + service, _, wallet := spoolTestService(t, store) + path := filepath.Join(t.TempDir(), "grants.json") + if err := service.AttachGrantSpoolPath(path); err != nil { + t.Fatal(err) + } + request := wire.AppendVarint(nil, 1, 1) + if _, _, _, err := service.Handle("/MailInfo", request); err != nil { + t.Fatal(err) + } + writeSpool(t, path, + spoolGrant("dev-mail:a", GrantReward{Type: 4, Count: 123}), + spoolGrant("dev-mail:b", GrantReward{Type: 8, ID: 9, Count: 10}), + ) + // Spool attachment never changes the versioned starter mailbox. + if len(service.Starter.Mails) != 0 { + t.Fatal("starter changed") + } + for i := 0; i < 2; i++ { + code, response, handled, err := service.Handle("/MailInfo", request) + if err != nil || !handled || code != packetCode { + t.Fatalf("code=%d handled=%v err=%v", code, handled, err) + } + count, _, _ := wire.Varint(response, 2) + max, _, _ := wire.Varint(response, 3) + if count != 3 || max != 102 { + t.Fatalf("count=%d max=%d", count, max) + } + } + if wallet.Snapshot().Gold != 0 { + t.Fatal("unopened mail granted currency") + } + open := wire.AppendVarint(wire.AppendVarint(nil, 1, 2), 2, 101) + if _, _, _, err := service.Handle("/MailOpen", open); err != nil { + t.Fatal(err) + } + reopened, inventory, wallet := spoolTestService(t, store) + if err := reopened.AttachGrantSpoolPath(path); err != nil { + t.Fatal(err) + } + _, response, _, err := reopened.Handle("/MailInfo", request) + if err != nil { + t.Fatal(err) + } + if count, _, _ := wire.Varint(response, 2); count != 2 { + t.Fatalf("count=%d", count) + } + if len(reopened.dynamic) != 2 || reopened.state.NextDynamicMailID != 103 { + t.Fatalf("reissued mail: %+v", reopened.state) + } + open = wire.AppendVarint(wire.AppendVarint(nil, 1, 3), 2, 102) + if _, _, _, err := reopened.Handle("/MailOpen", open); err != nil { + t.Fatal(err) + } + if wallet.Snapshot().Gold != 123 || len(inventory.All()) != 1 || inventory.All()[0].Count != 10 { + t.Fatalf("wallet=%+v inventory=%+v", wallet.Snapshot(), inventory.All()) + } + if _, _, _, err := reopened.Handle("/MailInfo", request); err != nil { + t.Fatal(err) + } + if reopened.state.NextDynamicMailID != 103 { + t.Fatal("opened grants were reissued") + } +} + +func TestGrantSpoolPersistsIssuedIdentityInSQLiteRequestTransaction(t *testing.T) { + dir := t.TempDir() + database := filepath.Join(dir, "state.db") + path := filepath.Join(dir, "grants.json") + writeSpool(t, path, spoolGrant("dev-mail:sqlite", GrantReward{Type: 4, Count: 123})) + repository, err := accountstate.Open(database) + if err != nil { + t.Fatal(err) + } + service, _, _ := spoolTestService(t, repository) + if err := service.AttachGrantSpoolPath(path); err != nil { + t.Fatal(err) + } + operation, err := repository.BeginOperation() + if err != nil { + t.Fatal(err) + } + request := wire.AppendVarint(nil, 1, 1) + if _, _, _, err := service.Handle("/MailInfo", request); err != nil { + t.Fatal(err) + } + if err := operation.Commit(); err != nil { + t.Fatal(err) + } + if err := repository.Close(); err != nil { + t.Fatal(err) + } + repository, err = accountstate.Open(database) + if err != nil { + t.Fatal(err) + } + defer repository.Close() + service, _, _ = spoolTestService(t, repository) + if err := service.AttachGrantSpoolPath(path); err != nil { + t.Fatal(err) + } + operation, err = repository.BeginOperation() + if err != nil { + t.Fatal(err) + } + if _, _, _, err := service.Handle("/MailInfo", request); err != nil { + t.Fatal(err) + } + if err := operation.Commit(); err != nil { + t.Fatal(err) + } + if len(service.dynamic) != 1 || service.issued["dev-mail:sqlite"] != 101 || service.state.NextDynamicMailID != 102 { + t.Fatalf("SQLite restart reissued grant: state=%+v issued=%+v", service.state, service.issued) + } +} + +func TestGrantSpoolValidatesEntireBatchBeforeIssuing(t *testing.T) { + store := stateio.NewMemory() + service, _, _ := spoolTestService(t, store) + path := filepath.Join(t.TempDir(), "grants.json") + if err := service.AttachGrantSpoolPath(path); err != nil { + t.Fatal(err) + } + good := spoolGrant("a", GrantReward{Type: 4, Count: 10}) + bad := spoolGrant("b", GrantReward{Type: 6, ID: 1, Count: 1}) + writeSpool(t, path, good, bad) + if _, _, _, err := service.Handle("/MailInfo", wire.AppendVarint(nil, 1, 1)); err == nil { + t.Fatal("accepted unsupported second grant") + } + assertNoSpoolMail(t, service, store) + writeSpool(t, path, good, good) + if _, _, _, err := service.Handle("/MailInfo", wire.AppendVarint(nil, 1, 2)); err == nil { + t.Fatal("accepted duplicate identity") + } + assertNoSpoolMail(t, service, store) + writeSpool(t, path, good) + if _, _, _, err := service.Handle("/MailInfo", wire.AppendVarint(nil, 1, 3)); err != nil { + t.Fatal(err) + } + if service.issued["a"] != 101 { + t.Fatal("valid replacement did not import") + } +} + +func assertNoSpoolMail(t *testing.T, service *Service, store stateio.EntryStore) { + t.Helper() + if len(service.dynamic) != 0 || len(service.issued) != 0 || service.state.NextDynamicMailID != 101 { + t.Fatalf("partial memory update: %+v", service.state) + } + for _, bucket := range []string{"dynamic", "issued"} { + entries, err := store.ListEntries("mail", bucket) + if err != nil || len(entries) != 0 { + t.Fatalf("partial storage %s: entries=%v err=%v", bucket, entries, err) + } + } +} + +type failingSpoolStore struct { + *stateio.Memory + fail bool + writes int +} + +func (s *failingSpoolStore) SaveWithEntries(domain string, core []byte, changes []stateio.EntryMutation) error { + if domain == "mail" && len(changes) != 0 { + s.writes++ + if s.fail { + return errors.New("injected spool save failure") + } + } + return s.Memory.SaveWithEntries(domain, core, changes) +} + +func TestGrantSpoolBatchWriteFailureLeavesMemoryAndStorageUnchanged(t *testing.T) { + store := &failingSpoolStore{Memory: stateio.NewMemory(), fail: true} + service, _, _ := spoolTestService(t, store) + path := filepath.Join(t.TempDir(), "grants.json") + writeSpool(t, path, spoolGrant("a", GrantReward{Type: 4, Count: 10}), spoolGrant("b", GrantReward{Type: 4, Count: 20})) + if err := service.AttachGrantSpoolPath(path); err != nil { + t.Fatal(err) + } + request := wire.AppendVarint(nil, 1, 1) + if _, _, _, err := service.Handle("/MailInfo", request); err == nil { + t.Fatal("save failure ignored") + } + assertNoSpoolMail(t, service, store) + if store.writes != 1 { + t.Fatalf("batch used %d writes", store.writes) + } + store.fail = false + if _, _, _, err := service.Handle("/MailInfo", request); err != nil { + t.Fatal(err) + } + if len(service.dynamic) != 2 || service.state.NextDynamicMailID != 103 { + t.Fatal("retry did not issue complete batch") + } +} + +func TestGrantSpoolBatchIDExhaustionDoesNotIssueFirstMail(t *testing.T) { + store := stateio.NewMemory() + service, _, _ := spoolTestService(t, store) + service.state.NextDynamicMailID = ^uint64(0) - 1 + path := filepath.Join(t.TempDir(), "grants.json") + writeSpool(t, path, spoolGrant("a", GrantReward{Type: 4, Count: 10}), spoolGrant("b", GrantReward{Type: 4, Count: 20})) + if err := service.AttachGrantSpoolPath(path); err != nil { + t.Fatal(err) + } + if _, _, _, err := service.Handle("/MailInfo", wire.AppendVarint(nil, 1, 1)); err == nil { + t.Fatal("accepted overflowing batch") + } + if len(service.dynamic) != 0 || service.state.NextDynamicMailID != ^uint64(0)-1 { + t.Fatal("partially imported overflowing batch") + } +} + +func TestGrantSpoolRejectsMalformedSchema(t *testing.T) { + valid := `{"version":1,"grants":[{"identity":"a","title":"t","body":"b","sent_at":1790989200000,"rewards":[{"type":4,"id":0,"count":10}]}]}` + if _, err := decodeGrantSpool([]byte(valid)); err != nil { + t.Fatal(err) + } + invalid := []string{ + `null`, `{}`, `{"version":1,"grants":null}`, `{"version":1,"grants":[],"extra":0}`, + strings.Replace(valid, `"version":1`, `"version":2`, 1), + strings.Replace(valid, `"version":1`, `"version":1,"version":1`, 1), + strings.Replace(valid, `"title":"t"`, `"title":"t","title":"u"`, 1), + strings.Replace(valid, `"title":"t"`, `"Title":"t"`, 1), + strings.Replace(valid, `"body":"b"`, `"body":null`, 1), + strings.Replace(valid, `"identity":"a"`, `"identity":" "`, 1), + strings.Replace(valid, `"sent_at":1790989200000`, `"sent_at":0`, 1), + strings.Replace(valid, `"sent_at":1790989200000`, `"sent_at":9223372036854775807`, 1), + strings.Replace(valid, `"count":10`, `"count":0`, 1), + strings.Replace(valid, `"count":10`, `"count":1.5`, 1), + strings.Replace(valid, `"count":10`, `"count":null`, 1), + strings.Replace(valid, `"count":10`, `"count":10,"count":10`, 1), + strings.Replace(valid, `"count":10`, `"count":10,"extra":0`, 1), + strings.Replace(valid, `"id":0,`, ``, 1), + strings.Replace(valid, `"id":0`, `"id":1`, 1), + strings.Replace(valid, `"type":4`, `"type":8`, 1), + valid + `{}`, + } + for _, input := range invalid { + if _, err := decodeGrantSpool([]byte(input)); err == nil { + t.Errorf("accepted malformed spool: %s", input) + } + } + if _, err := decodeGrantSpool([]byte(`{"version":1,"grants":[]}`)); err != nil { + t.Fatal(err) + } +} diff --git a/go/internal/server/mail/mail.go b/go/internal/server/mail/mail.go index 77b7c53..1aaf368 100644 --- a/go/internal/server/mail/mail.go +++ b/go/internal/server/mail/mail.go @@ -197,16 +197,17 @@ type stateSnapshot struct { // Service owns mailbox visibility and idempotent reward delivery. Starter is // immutable source data; only opened IDs are persisted in the player state. type Service struct { - mu sync.Mutex - Starter *Starter - seedPath string - seedStamp fileStamp - storage stateio.AtomicEntryStore - inventory *player.Inventory - wallet *player.Wallet - state stateSnapshot - dynamic map[uint64]MailDBInfo - issued map[string]uint64 + mu sync.Mutex + Starter *Starter + seedPath string + seedStamp fileStamp + grantSpoolPath string + storage stateio.AtomicEntryStore + inventory *player.Inventory + wallet *player.Wallet + state stateSnapshot + dynamic map[uint64]MailDBInfo + issued map[string]uint64 } type fileStamp struct { @@ -339,9 +340,16 @@ func (s *Service) Handle(path string, request []byte) (int, []byte, bool, error) s.mu.Lock() defer s.mu.Unlock() if path == "/MailInfo" { + grants, err := readGrantSpool(s.grantSpoolPath) + if err != nil { + return 0, nil, true, err + } if err := s.reloadSeedIfChanged(); err != nil { return 0, nil, true, err } + if err := s.enqueueCompensations(grants); err != nil { + return 0, nil, true, err + } return packetCode, s.info(), true, nil } response, err := s.open(request) @@ -571,54 +579,72 @@ func (s *Service) persist(next stateSnapshot) error { // completed reward. Identity is period-scoped and makes repeated rollover // checks idempotent. func (s *Service) EnqueueCompensation(identity, title, body string, rewards []gamedata.Reward, sentAt time.Time) error { - if identity == "" || title == "" || body == "" || sentAt.IsZero() || len(rewards) == 0 { - return errors.New("mail: invalid compensation") + grant := compensation{identity: identity, title: title, body: body, rewards: rewards, sentAt: sentAt} + if err := grant.validate(); err != nil { + return err } s.mu.Lock() defer s.mu.Unlock() - if _, exists := s.issued[identity]; exists { - return nil - } - if s.state.NextDynamicMailID == 0 || s.state.NextDynamicMailID == ^uint64(0) { - return errors.New("mail: dynamic mail ID exhausted") - } - entry := MailDBInfo{ - MailID: s.state.NextDynamicMailID, MailType: 2, Title: title, Body: body, - SentAt: uint64(sentAt.UTC().UnixMilli()), ExpiresAt: uint64(sentAt.UTC().Add(30 * 24 * time.Hour).UnixMilli()), - } - for _, reward := range rewards { - if reward.Type == 0 || reward.Count == 0 || (!currencyRewardTypes[reward.Type] && (!itemDBInfoTypes[reward.Type] || reward.ID == 0)) { - return fmt.Errorf("mail: compensation %q has unsupported reward type=%d id=%d count=%d", identity, reward.Type, reward.ID, reward.Count) - } - entry.RewardTypes = append(entry.RewardTypes, reward.Type) - entry.RewardIDs = append(entry.RewardIDs, reward.ID) - entry.RewardCounts = append(entry.RewardCounts, reward.Count) - } - payload, err := json.Marshal(entry) - if err != nil { - return err - } - issuedPayload, err := json.Marshal(entry.MailID) - if err != nil { - return err - } + return s.enqueueCompensations([]compensation{grant}) +} + +// enqueueCompensations shares the EnqueueCompensation allocator and durable +// identity ledger. The entire batch is prepared before a single atomic write; +// neither storage nor memory can retain a partially imported grant spool. +// Caller holds s.mu and has validated every grant. +func (s *Service) enqueueCompensations(grants []compensation) error { next := s.state next.Opened = append([]uint64(nil), s.state.Opened...) - next.NextDynamicMailID++ + var changes []stateio.EntryMutation + var entries []MailDBInfo + var identities []string + for _, grant := range grants { + if _, exists := s.issued[grant.identity]; exists { + continue + } + if next.NextDynamicMailID == 0 || next.NextDynamicMailID == ^uint64(0) { + return errors.New("mail: dynamic mail ID exhausted") + } + entry := MailDBInfo{ + MailID: next.NextDynamicMailID, MailType: 2, Title: grant.title, Body: grant.body, + SentAt: uint64(grant.sentAt.UTC().UnixMilli()), ExpiresAt: uint64(grant.sentAt.UTC().Add(30 * 24 * time.Hour).UnixMilli()), + } + for _, reward := range grant.rewards { + entry.RewardTypes = append(entry.RewardTypes, reward.Type) + entry.RewardIDs = append(entry.RewardIDs, reward.ID) + entry.RewardCounts = append(entry.RewardCounts, reward.Count) + } + payload, err := json.Marshal(entry) + if err != nil { + return err + } + issuedPayload, err := json.Marshal(entry.MailID) + if err != nil { + return err + } + changes = append(changes, + stateio.EntryMutation{Bucket: "dynamic", Key: strconv.FormatUint(entry.MailID, 10), Payload: payload}, + stateio.EntryMutation{Bucket: "issued", Key: grant.identity, Payload: issuedPayload}, + ) + entries = append(entries, entry) + identities = append(identities, grant.identity) + next.NextDynamicMailID++ + } + if len(entries) == 0 { + return nil + } core, err := json.Marshal(next) if err != nil { return err } - changes := []stateio.EntryMutation{ - {Bucket: "dynamic", Key: strconv.FormatUint(entry.MailID, 10), Payload: payload}, - {Bucket: "issued", Key: identity, Payload: issuedPayload}, - } if err := s.storage.SaveWithEntries("mail", core, changes); err != nil { return err } s.state = next - s.dynamic[entry.MailID] = entry - s.issued[identity] = entry.MailID + for i, entry := range entries { + s.dynamic[entry.MailID] = entry + s.issued[identities[i]] = entry.MailID + } return nil } diff --git a/tools/python/dev_mail_grant.py b/tools/python/dev_mail_grant.py index f27ff7b..7e1e2cc 100644 --- a/tools/python/dev_mail_grant.py +++ b/tools/python/dev_mail_grant.py @@ -1,5 +1,5 @@ #!/usr/bin/env python3 -"""Local-only BD2 development browser tool. +"""Local-only BD2 development mail and browser tool. It provides development mail grants and loopback-only runtime settings without reading or changing account state. Files are replaced atomically and consumed @@ -9,11 +9,17 @@ Example: python tools/python/dev_mail_grant.py serve ` --game-data E:\\bd2\\dl\\GameData --game-data-version 20260923193640 ` --mail-seed go\\seed\\v2_35_10\\mail.json --output data\\dev\\mail-grants.json + + python tools/python/dev_mail_grant.py grant ` + --output data\\dev\\currency-grants.json --identity test-grant-1 ` + --attachment 4:0:10000 --attachment 3:0:100 """ from __future__ import annotations import argparse +from contextlib import contextmanager +import errno import html import json import os @@ -23,13 +29,19 @@ import sys import tempfile import threading import time +import uuid from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from typing import Any -# gamedata_db is the repository's reviewed, read-only GameData decryptor. +# The grant command requires only the Python standard library. GameData's +# optional decryptor dependency is imported only by the browser's data readers. sys.path.insert(0, str(Path(__file__).resolve().parent)) -from gamedata_db import read_database, walk_wire # noqa: E402 + + +def _gamedata(): + import gamedata_db + return gamedata_db VERSION = "2.35.10" @@ -65,7 +77,7 @@ def _varint(value: Any) -> int: def fields(proto: bytes) -> dict[int, list[Any]]: result: dict[int, list[Any]] = {} - for number, wire_type, value in walk_wire(proto): + for number, wire_type, value in _gamedata().walk_wire(proto): if wire_type != 0 and wire_type != 2: continue result.setdefault(number, []).append(value) @@ -114,7 +126,7 @@ def packed_varints(values: dict[int, list[Any]], number: int) -> list[int]: def open_readonly_database(root: Path, version: str) -> tuple[sqlite3.Connection, Path]: """Open the current common database in a private read-only SQLite file.""" - plain = read_database(root, version, "quest") + plain = _gamedata().read_database(root, version, "quest") handle = tempfile.NamedTemporaryFile(prefix="bd2-dev-mail-", suffix=".db", delete=False) path = Path(handle.name) try: @@ -399,9 +411,10 @@ def normalise_seed(seed: dict[str, Any]) -> dict[str, Any]: def atomic_json(path: Path, value: dict[str, Any]) -> None: path = path.resolve() path.parent.mkdir(parents=True, exist_ok=True) - temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp") + descriptor, name = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=path.parent) + temporary = Path(name) try: - with temporary.open("w", encoding="utf-8", newline="\n") as stream: + with os.fdopen(descriptor, "w", encoding="utf-8", newline="\n") as stream: json.dump(value, stream, ensure_ascii=False, indent=2) stream.write("\n") stream.flush() @@ -411,6 +424,124 @@ def atomic_json(path: Path, value: dict[str, Any]) -> None: temporary.unlink(missing_ok=True) +@contextmanager +def grant_file_lock(path: Path): + """Serialize each complete read/append/replace across CLI processes. + + Keep the sidecar lock file: deleting it would allow another process to lock + a different file while a waiting process still owns the original inode. + """ + path = path.resolve() + path.parent.mkdir(parents=True, exist_ok=True) + with path.with_name(path.name + ".lock").open("a+b") as stream: + if os.name == "nt": + import msvcrt + stream.seek(0, os.SEEK_END) + if stream.tell() == 0: + stream.write(b"\0") + stream.flush() + while True: + stream.seek(0) + try: + msvcrt.locking(stream.fileno(), msvcrt.LK_NBLCK, 1) + break + except OSError as exc: + if exc.errno not in {errno.EACCES, errno.EAGAIN, errno.EDEADLK}: + raise + time.sleep(0.05) + try: + yield + finally: + stream.seek(0) + msvcrt.locking(stream.fileno(), msvcrt.LK_UNLCK, 1) + else: + import fcntl + fcntl.flock(stream.fileno(), fcntl.LOCK_EX) + try: + yield + finally: + fcntl.flock(stream.fileno(), fcntl.LOCK_UN) + + +def _validate_reward(value: Any) -> dict[str, int]: + if not isinstance(value, dict) or set(value) != {"type", "id", "count"}: + raise ValueError("附件必须只包含 type、id、count") + if type(value["type"]) is not int or value["type"] not in MAIL_CURRENCY_TYPES: + raise ValueError("附件 type 只允许 3、4、12、20") + if type(value["id"]) is not int or value["id"] != 0: + raise ValueError("货币附件 id 必须为 0") + if type(value["count"]) is not int or not 1 <= value["count"] <= MAX_INT32: + raise ValueError(f"附件 count 必须是 1 到 {MAX_INT32} 的整数") + return dict(value) + + +def attachment(value: str) -> dict[str, int]: + try: + parts = value.split(":") + if len(parts) != 3: + raise ValueError("附件格式必须是 TYPE:ID:COUNT") + return _validate_reward(dict(zip(("type", "id", "count"), map(int, parts)))) + except ValueError as exc: + raise argparse.ArgumentTypeError(str(exc)) from exc + + +def _validate_grant(value: Any) -> dict[str, Any]: + if not isinstance(value, dict) or set(value) != {"identity", "title", "body", "sent_at", "rewards"}: + raise ValueError("发放记录必须只包含 identity、title、body、sent_at、rewards") + identity, title, body = value["identity"], value["title"], value["body"] + if not isinstance(identity, str) or not identity.strip() or len(identity) > 500: + raise ValueError("identity 不能为空且不超过 500 字符") + if not isinstance(title, str) or not title.strip() or len(title) > 500: + raise ValueError("标题不能为空且不超过 500 字符") + if not isinstance(body, str) or not body.strip() or len(body) > 5000: + raise ValueError("正文不能为空且不超过 5000 字符") + if type(value["sent_at"]) is not int or not 1 <= value["sent_at"] <= (1 << 63) - 1: + raise ValueError("sent_at 必须是正 int64 毫秒时间戳") + if not isinstance(value["rewards"], list) or not value["rewards"]: + raise ValueError("至少需要一个附件") + return {**value, "rewards": [_validate_reward(reward) for reward in value["rewards"]]} + + +def load_grants(path: Path) -> dict[str, Any]: + if not path.exists(): + return {"version": 1, "grants": []} + try: + value = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise ValueError(f"无法读取邮件发放文件 {path}: {exc}") from exc + if not isinstance(value, dict) or set(value) != {"version", "grants"} or type(value["version"]) is not int or value["version"] != 1 or not isinstance(value["grants"], list): + raise ValueError("邮件发放文件必须只包含 version=1 和 grants 数组") + grants = [_validate_grant(entry) for entry in value["grants"]] + identities = [entry["identity"] for entry in grants] + if len(set(identities)) != len(identities): + raise ValueError("邮件发放文件包含重复 identity") + return {"version": 1, "grants": grants} + + +def grant(args: argparse.Namespace) -> int: + entry = _validate_grant({ + "identity": args.identity.strip() if args.identity is not None else str(uuid.uuid4()), + "title": args.title.strip(), + "body": args.body.strip(), + "sent_at": time.time_ns() // 1_000_000, + "rewards": args.attachment, + }) + output = args.output.resolve() + with grant_file_lock(output): + value = load_grants(output) + existing = next((item for item in value["grants"] if item["identity"] == entry["identity"]), None) + if existing is not None: + if any(existing[key] != entry[key] for key in ("title", "body", "rewards")): + raise ValueError(f"identity {entry['identity']!r} 已存在且内容不同") + entry, created = existing, False + else: + value["grants"].append(entry) + atomic_json(output, value) + created = True + print(json.dumps({"grant": entry, "output": str(output), "created": created}, ensure_ascii=False)) + return 0 + + class MailGrantStore: def __init__(self, source: Path, output: Path, items: list[dict[str, Any]], expires_days: int): self.source = source.resolve() @@ -623,13 +754,20 @@ def parser() -> argparse.ArgumentParser: command.add_argument("--listen-port", default=8765, type=int, help="loopback port (default: 8765)") command.add_argument("--expires-days", default=365, type=int, help="development mail validity (default: 365)") command.set_defaults(run=serve) + command = commands.add_parser("grant", help="append one durable currency mail grant (standard library only)") + command.add_argument("--output", type=Path, required=True, help="version=1 development mail grants JSON") + command.add_argument("--attachment", type=attachment, action="append", required=True, metavar="TYPE:ID:COUNT", help="currency reward; repeat to include multiple attachments in one mail") + command.add_argument("--identity", help="stable idempotency identity (default: a new UUID)") + command.add_argument("--title", default="开发测试物品", help="mail title (maximum 500 characters)") + command.add_argument("--body", default="由本地开发邮件工具发放。", help="mail body (maximum 5000 characters)") + command.set_defaults(run=grant) return result def main() -> int: args = parser().parse_args() try: - if args.listen_host not in {"127.0.0.1", "localhost", "::1"}: + if args.command == "serve" and args.listen_host not in {"127.0.0.1", "localhost", "::1"}: raise ValueError("开发工具只允许监听本机回环地址") return args.run(args) except (OSError, ValueError, sqlite3.Error) as exc: diff --git a/tools/python/tests/test_dev_mail_grant.py b/tools/python/tests/test_dev_mail_grant.py new file mode 100644 index 0000000..e7413cc --- /dev/null +++ b/tools/python/tests/test_dev_mail_grant.py @@ -0,0 +1,178 @@ +"""Currency grant CLI regressions; runnable without third-party packages.""" + +from __future__ import annotations + +from contextlib import redirect_stdout +import io +import json +from pathlib import Path +import subprocess +import sys +import tempfile +import unittest +from unittest import mock + +TOOLS = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(TOOLS)) +import dev_mail_grant + + +class CurrencyMailGrantTests(unittest.TestCase): + def setUp(self): + self.temporary = tempfile.TemporaryDirectory() + self.addCleanup(self.temporary.cleanup) + self.output = Path(self.temporary.name) / "nested" / "grants.json" + + def args(self, *extra): + return dev_mail_grant.parser().parse_args([ + "grant", "--output", str(self.output), "--attachment", "4:0:123", *extra, + ]) + + def run_grant(self, *extra): + stream = io.StringIO() + with redirect_stdout(stream): + self.assertEqual(dev_mail_grant.grant(self.args(*extra)), 0) + return json.loads(stream.getvalue()) + + def child(self, *extra): + # -S removes site packages, proving the command can run without Crypto. + return [sys.executable, "-S", str(TOOLS / "dev_mail_grant.py"), + "grant", "--output", str(self.output), "--attachment", "4:0:123", *extra] + + def test_multiple_attachments_create_one_mail_and_preserve_duplicates(self): + result = self.run_grant("--identity", "release-1", "--title", " 发放测试 ", + "--body", " 正文 ", "--attachment", "3:0:5", + "--attachment", "4:0:2", "--attachment", "12:0:1", + "--attachment", "20:0:2147483647") + value = dev_mail_grant.load_grants(self.output) + self.assertEqual(value["version"], 1) + self.assertEqual(len(value["grants"]), 1) + self.assertEqual(value["grants"][0], result["grant"]) + self.assertEqual(result["grant"]["title"], "发放测试") + self.assertEqual(result["grant"]["body"], "正文") + self.assertGreater(result["grant"]["sent_at"], 0) + self.assertEqual(result["grant"]["rewards"], [ + {"type": 4, "id": 0, "count": 123}, {"type": 3, "id": 0, "count": 5}, + {"type": 4, "id": 0, "count": 2}, {"type": 12, "id": 0, "count": 1}, + {"type": 20, "id": 0, "count": 2147483647}, + ]) + + def test_identity_retry_keeps_original_timestamp_and_does_not_write(self): + first = self.run_grant("--identity", "once") + original = self.output.read_bytes() + with mock.patch.object(dev_mail_grant, "atomic_json", side_effect=AssertionError("must not write")): + second = self.run_grant("--identity", "once") + self.assertTrue(first["created"]) + self.assertFalse(second["created"]) + self.assertEqual(first["grant"], second["grant"]) + self.assertEqual(self.output.read_bytes(), original) + + def test_identity_conflict_is_rejected_without_changing_file(self): + self.run_grant("--identity", "once") + original = self.output.read_bytes() + for extra in (("--title", "another"), ("--body", "another"), ("--attachment", "3:0:1")): + with self.subTest(extra=extra), self.assertRaisesRegex(ValueError, "内容不同"): + self.run_grant("--identity", "once", *extra) + self.assertEqual(self.output.read_bytes(), original) + + def test_omitted_identity_appends_distinct_grants(self): + first, second = self.run_grant(), self.run_grant() + self.assertNotEqual(first["grant"]["identity"], second["grant"]["identity"]) + self.assertEqual(len(dev_mail_grant.load_grants(self.output)["grants"]), 2) + + def test_invalid_attachments_are_rejected(self): + for attachment in ("8:1:1", "4:1:1", "4:0:0", "4:0:-1", "4:0:2147483648", "4:0", "4:0:1:2", "4:0:true"): + with self.subTest(attachment=attachment), self.assertRaises(SystemExit), redirect_stdout(io.StringIO()): + with mock.patch("sys.stderr", new=io.StringIO()): + self.args("--attachment", attachment) + self.assertFalse(self.output.exists()) + + def test_invalid_title_body_or_identity_does_not_create_output(self): + for extra in (("--title", " "), ("--title", "x" * 501), ("--body", "x" * 5001), + ("--identity", " "), ("--identity", "x" * 501)): + with self.subTest(extra=extra), self.assertRaises(ValueError): + self.run_grant(*extra) + self.assertFalse(self.output.exists()) + + def test_existing_invalid_schema_is_preserved(self): + entry = {"identity": "one", "title": "title", "body": "body", "sent_at": 1, + "rewards": [{"type": 4, "id": 0, "count": 1}]} + invalid_values = [ + [], {"version": True, "grants": []}, {"version": 1, "grants": [], "extra": 1}, + {"version": "2.35.10", "mails": []}, {"version": 1, "grants": [entry, entry]}, + {"version": 1, "grants": [{**entry, "sent_at": True}]}, + {"version": 1, "grants": [{**entry, "extra": 1}]}, + {"version": 1, "grants": [{**entry, "rewards": []}]}, + {"version": 1, "grants": [{**entry, "rewards": [{"type": 4, "id": False, "count": 1}]}]}, + {"version": 1, "grants": [{**entry, "rewards": [{"type": 4, "id": 0, "count": True}]}]}, + ] + self.output.parent.mkdir(parents=True) + for value in invalid_values: + original = json.dumps(value).encode() + self.output.write_bytes(original) + with self.subTest(value=value), self.assertRaises(ValueError): + self.run_grant() + self.assertEqual(self.output.read_bytes(), original) + self.output.write_text("broken JSON", encoding="utf-8") + with self.assertRaises(ValueError): + self.run_grant() + self.assertEqual(self.output.read_text(encoding="utf-8"), "broken JSON") + + def test_failed_atomic_replace_preserves_previous_file_and_cleans_temporary(self): + self.run_grant("--identity", "one") + original = self.output.read_bytes() + with mock.patch.object(dev_mail_grant.os, "replace", side_effect=OSError("injected failure")): + with self.assertRaises(OSError): + self.run_grant("--identity", "two") + self.assertEqual(self.output.read_bytes(), original) + self.assertEqual(list(self.output.parent.glob("*.tmp")), []) + + def test_cli_uses_standard_library_and_reports_validation_errors(self): + result = subprocess.run(self.child("--identity", "standalone"), capture_output=True, text=True, timeout=20) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertTrue(json.loads(result.stdout)["created"]) + result = subprocess.run(self.child("--identity", "standalone", "--title", "conflict"), capture_output=True, text=True, timeout=20) + self.assertEqual(result.returncode, 1) + self.assertIn("dev_tools:", result.stderr) + + def test_process_waits_for_lock_then_reads_latest_file(self): + with dev_mail_grant.grant_file_lock(self.output): + process = subprocess.Popen(self.child("--identity", "child"), stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) + try: + with self.assertRaises(subprocess.TimeoutExpired): + process.wait(timeout=0.5) + entry = {"identity": "parent", "title": "title", "body": "body", "sent_at": 1, + "rewards": [{"type": 4, "id": 0, "count": 1}]} + dev_mail_grant.atomic_json(self.output, {"version": 1, "grants": [entry]}) + except BaseException: + process.kill() + process.communicate() + raise + stdout, stderr = process.communicate(timeout=20) + self.assertEqual(process.returncode, 0, stderr) + self.assertTrue(json.loads(stdout)["created"]) + self.assertEqual([item["identity"] for item in dev_mail_grant.load_grants(self.output)["grants"]], ["parent", "child"]) + + def test_parallel_processes_preserve_all_grants_and_shared_identity_once(self): + identities = [f"child-{i}" for i in range(8)] + ["shared"] * 4 + processes = [subprocess.Popen(self.child("--identity", identity), stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) + for identity in identities] + try: + results = [] + for process in processes: + stdout, stderr = process.communicate(timeout=30) + self.assertEqual(process.returncode, 0, stderr) + results.append(json.loads(stdout)) + grants = dev_mail_grant.load_grants(self.output)["grants"] + self.assertEqual({entry["identity"] for entry in grants}, set(identities)) + self.assertEqual(len(grants), 9) + self.assertEqual(sum(result["created"] for result in results), 9) + finally: + for process in processes: + if process.poll() is None: + process.kill() + process.communicate() + + +if __name__ == "__main__": + unittest.main()