Compare commits
No commits in common. "a4443cbe4c873994a077c23ce131c0bcb3e79545" and "e0a3410f36d71a86d9e32c5e5d8e5ef10b32f7a6" have entirely different histories.
a4443cbe4c
...
e0a3410f36
@ -28,7 +28,7 @@ func asCustodialRegistrationEvent(gev *geEvent.Event) (*eventCustodialRegistrati
|
||||
if !ok {
|
||||
return nil, false
|
||||
}
|
||||
logg.Debugf("parsed ev", "pl", pl, "ev", ev)
|
||||
logg.Debug("parsed ev", "pl", pl, "ev", ev)
|
||||
return &ev, true
|
||||
}
|
||||
|
||||
|
@ -3,18 +3,19 @@ package nats
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"log/slog"
|
||||
"os"
|
||||
|
||||
nats "github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
geEvent "github.com/grassrootseconomics/eth-tracker/pkg/event"
|
||||
"git.defalsify.org/vise.git/logging"
|
||||
"git.defalsify.org/vise.git/db"
|
||||
"git.grassecon.net/urdt/ussd/common"
|
||||
"git.grassecon.net/term/event"
|
||||
)
|
||||
|
||||
var (
|
||||
logg = logging.NewVanilla().WithDomain("term-nats")
|
||||
logg = slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelDebug}))
|
||||
)
|
||||
|
||||
type NatsSubscription struct {
|
||||
@ -77,30 +78,30 @@ func(n *NatsSubscription) Close() error {
|
||||
func fail(m jetstream.Msg) {
|
||||
err := m.Nak()
|
||||
if err != nil {
|
||||
logg.Errorf("nats nak fail", "err", err)
|
||||
logg.Error("nats nak fail", "err", err)
|
||||
}
|
||||
}
|
||||
|
||||
func(n *NatsSubscription) handleEvent(m jetstream.Msg) {
|
||||
var ev geEvent.Event
|
||||
|
||||
logg.DebugCtxf(n.ctx, "have msg", "err", m)
|
||||
logg.Debug("have msg", "err", m)
|
||||
b := m.Data()
|
||||
err := json.Unmarshal(b, &ev)
|
||||
if err != nil {
|
||||
logg.ErrorCtxf(n.ctx, "nats msg deserialize fail", "err", err)
|
||||
logg.Error("nats msg deserialize fail", "err", err)
|
||||
//fail(m)
|
||||
} else {
|
||||
err = n.Route(n.ctx, &ev)
|
||||
if err != nil {
|
||||
logg.ErrorCtxf(n.ctx, "handler route fail", "err", err)
|
||||
logg.Error("handler route fail", "err", err)
|
||||
//fail(m)
|
||||
}
|
||||
}
|
||||
err = m.Ack()
|
||||
if err != nil {
|
||||
logg.ErrorCtxf(n.ctx, "ack fail", "err", err)
|
||||
logg.Error("ack fail", "err", err)
|
||||
panic("ack fail")
|
||||
}
|
||||
logg.DebugCtxf(n.ctx, "handle msg complete")
|
||||
logg.Debug("handle msg complete")
|
||||
}
|
||||
|
@ -1,166 +0,0 @@
|
||||
package nats
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
nats "github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
dataserviceapi "github.com/grassrootseconomics/ussd-data-service/pkg/api"
|
||||
"git.defalsify.org/vise.git/db"
|
||||
memdb "git.defalsify.org/vise.git/db/mem"
|
||||
"git.grassecon.net/urdt/ussd/common"
|
||||
"git.grassecon.net/urdt/ussd/config"
|
||||
"git.grassecon.net/urdt/ussd/models"
|
||||
"git.grassecon.net/term/lookup"
|
||||
)
|
||||
|
||||
func init() {
|
||||
}
|
||||
|
||||
const (
|
||||
aliceChecksum = "0xeae046BF396e91f5A8D74f863dC57c107c8a4a70"
|
||||
)
|
||||
|
||||
// TODO: jetstream, would have been nice of you to provide an easier way to make a mock msg
|
||||
type testMsg struct {
|
||||
data []byte
|
||||
}
|
||||
|
||||
func(m *testMsg) Ack() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func(m *testMsg) Nak() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func(m *testMsg) NakWithDelay(time.Duration) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func(m *testMsg) Data() []byte {
|
||||
return m.data
|
||||
}
|
||||
|
||||
func(m *testMsg) Reply() string {
|
||||
return ""
|
||||
}
|
||||
|
||||
func(m *testMsg) Subject() string {
|
||||
return ""
|
||||
}
|
||||
|
||||
func(m *testMsg) Term() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func(m *testMsg) TermWithReason(string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func(m *testMsg) DoubleAck(ctx context.Context) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func(m *testMsg) Headers() nats.Header {
|
||||
return nats.Header{}
|
||||
}
|
||||
|
||||
func(m *testMsg) InProgress() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func(m *testMsg) Metadata() (*jetstream.MsgMetadata, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
type mockApi struct {
|
||||
}
|
||||
|
||||
func(m mockApi) CheckBalance(ctx context.Context, publicKey string) (*models.BalanceResult, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func(m mockApi) CreateAccount(ctx context.Context) (*models.AccountResult, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func(m mockApi) TrackAccountStatus(ctx context.Context, publicKey string) (*models.TrackStatusResult, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func(m mockApi) FetchVouchers(ctx context.Context, publicKey string) ([]dataserviceapi.TokenHoldings, error) {
|
||||
logg.DebugCtxf(ctx, "mockapi fetchvouchers", "key", publicKey)
|
||||
return []dataserviceapi.TokenHoldings{
|
||||
dataserviceapi.TokenHoldings{
|
||||
ContractAddress: "0xeE0A29AE1BB7a033c8277C04780c4aBcf4388E93",
|
||||
TokenSymbol: "FOO",
|
||||
TokenDecimals: "6",
|
||||
Balance: "362436",
|
||||
},
|
||||
}, nil
|
||||
}
|
||||
|
||||
func(m mockApi) FetchTransactions(ctx context.Context, publicKey string) ([]dataserviceapi.Last10TxResponse, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func(m mockApi) VoucherData(ctx context.Context, address string) (*models.VoucherDataResult, error) {
|
||||
return &models.VoucherDataResult{
|
||||
TokenSymbol: "FOO",
|
||||
TokenName: "Foo Token",
|
||||
TokenDecimals: "6",
|
||||
SinkAddress: "0xb42C5920014eE152F2225285219407938469BBfA",
|
||||
}, nil
|
||||
}
|
||||
|
||||
func TestHandleMsg(t *testing.T) {
|
||||
err := config.LoadConfig()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
lookup.Api = mockApi{}
|
||||
|
||||
ctx := context.Background()
|
||||
userDb := memdb.NewMemDb()
|
||||
err = userDb.Connect(ctx, "")
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
alice, err := common.NormalizeHex(aliceChecksum)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
aliceSession := "5553425"
|
||||
userDb.SetSession(alice)
|
||||
userDb.SetPrefix(db.DATATYPE_USERDATA)
|
||||
err = userDb.Put(ctx, common.PackKey(common.DATA_PUBLIC_KEY_REVERSE, []byte{}), []byte(aliceSession))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
sub := NewNatsSubscription(userDb)
|
||||
|
||||
data := `{
|
||||
"block": 42,
|
||||
"contractAddress": "0x765DE816845861e75A25fCA122bb6898B8B1282a",
|
||||
"success": true,
|
||||
"timestamp": 1730592500,
|
||||
"transactionHash": "0xabcdef0123456789abcdef0123456789abcdef0123456789abcdef0123456789",
|
||||
"transactionType": "TOKEN_TRANSFER",
|
||||
"payload": {
|
||||
"from": "0xeae046BF396e91f5A8D74f863dC57c107c8a4a70",
|
||||
"to": "B3117202371853e24B725d4169D87616A7dDb127",
|
||||
"value": "1337"
|
||||
}
|
||||
}`
|
||||
msg := &testMsg{
|
||||
data: []byte(data),
|
||||
}
|
||||
sub.handleEvent(msg)
|
||||
}
|
@ -3,10 +3,11 @@ package event
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"os"
|
||||
|
||||
geEvent "github.com/grassrootseconomics/eth-tracker/pkg/event"
|
||||
|
||||
"git.defalsify.org/vise.git/logging"
|
||||
"git.grassecon.net/urdt/ussd/common"
|
||||
)
|
||||
|
||||
@ -16,7 +17,7 @@ const (
|
||||
)
|
||||
|
||||
var (
|
||||
logg = logging.NewVanilla().WithDomain("term-event")
|
||||
logg = slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelDebug}))
|
||||
)
|
||||
|
||||
type Router struct {
|
||||
@ -24,15 +25,10 @@ type Router struct {
|
||||
}
|
||||
|
||||
func(r *Router) Route(ctx context.Context, gev *geEvent.Event) error {
|
||||
logg.DebugCtxf(ctx, "have event", "ev", gev)
|
||||
logg.Debug("have event", "ev", gev)
|
||||
evCC, ok := asCustodialRegistrationEvent(gev)
|
||||
if ok {
|
||||
return handleCustodialRegistration(ctx, r.Store, evCC)
|
||||
}
|
||||
evTT, ok := asTokenTransferEvent(gev)
|
||||
if ok {
|
||||
return handleTokenTransfer(ctx, r.Store, evTT)
|
||||
}
|
||||
|
||||
return fmt.Errorf("unexpected message")
|
||||
}
|
||||
|
@ -10,6 +10,7 @@ import (
|
||||
|
||||
"git.defalsify.org/vise.git/db"
|
||||
"git.grassecon.net/urdt/ussd/common"
|
||||
"git.grassecon.net/urdt/ussd/remote"
|
||||
"git.grassecon.net/term/lookup"
|
||||
)
|
||||
|
||||
@ -28,13 +29,12 @@ type eventTokenTransfer struct {
|
||||
To string
|
||||
Value int
|
||||
TxHash string
|
||||
VoucherAddress string
|
||||
}
|
||||
|
||||
func updateTokenTransferList(ctx context.Context, store common.UserDataStore, identity lookup.Identity) error {
|
||||
func updateTokenTransferList(ctx context.Context, api remote.AccountServiceInterface, store common.UserDataStore, identity lookup.Identity) error {
|
||||
var r []string
|
||||
|
||||
txs, err := lookup.Api.FetchTransactions(ctx, identity.ChecksumAddress)
|
||||
txs, err := api.FetchTransactions(ctx, identity.ChecksumAddress)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@ -47,8 +47,8 @@ func updateTokenTransferList(ctx context.Context, store common.UserDataStore, id
|
||||
return store.WriteEntry(ctx, identity.SessionId, common.DATA_TRANSACTIONS, []byte(s))
|
||||
}
|
||||
|
||||
func updateTokenList(ctx context.Context, store *common.UserDataStore, identity lookup.Identity) error {
|
||||
holdings, err := lookup.Api.FetchVouchers(ctx, identity.ChecksumAddress)
|
||||
func updateTokenList(ctx context.Context, api remote.AccountServiceInterface, store *common.UserDataStore, identity lookup.Identity) error {
|
||||
holdings, err := api.FetchVouchers(ctx, identity.ChecksumAddress)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@ -57,8 +57,6 @@ func updateTokenList(ctx context.Context, store *common.UserDataStore, identity
|
||||
|
||||
// TODO: export subprefixdb and use that instead
|
||||
// TODO: make sure subprefixdb is thread safe when using gdbm
|
||||
// TODO: why is address session here unless explicitly set
|
||||
store.Db.SetSession(identity.SessionId)
|
||||
store.Db.SetPrefix(DATATYPE_USERSUB)
|
||||
|
||||
k := append([]byte("vouchers"), []byte("sym")...)
|
||||
@ -66,7 +64,7 @@ func updateTokenList(ctx context.Context, store *common.UserDataStore, identity
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
logg.TraceCtxf(ctx, "processvoucher", "key", k)
|
||||
|
||||
k = append([]byte("vouchers"), []byte("bal")...)
|
||||
err = store.Db.Put(ctx, k, []byte(metadata.Balances))
|
||||
if err != nil {
|
||||
@ -88,50 +86,44 @@ func updateTokenList(ctx context.Context, store *common.UserDataStore, identity
|
||||
return nil
|
||||
}
|
||||
|
||||
//func updateTokenBalance(ctx context.Context, api remote.AccountServiceInterface, store common.UserDataStore, sessionId string) error {
|
||||
// r, err := api.CheckBalance(ctx, sessionId)
|
||||
// if err != nil {
|
||||
// return err
|
||||
// }
|
||||
// //store.WriteEntry()
|
||||
// return nil
|
||||
//}
|
||||
//
|
||||
|
||||
func updateDefaultToken(ctx context.Context, store *common.UserDataStore, identity lookup.Identity, activeSym string) error {
|
||||
pfxDb := common.StoreToPrefixDb(store, []byte("vouchers"))
|
||||
// TODO: the activeSym input should instead be newline separated list?
|
||||
tokenData, err := common.GetVoucherData(ctx, pfxDb, activeSym)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
logg.TraceCtxf(ctx, "tokendaa", "d", tokenData)
|
||||
return common.UpdateVoucherData(ctx, store, identity.SessionId, tokenData)
|
||||
}
|
||||
|
||||
func updateWait(ctx context.Context) error {
|
||||
func updateWait(ctx context.Context, api remote.AccountServiceInterface) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func toSym(ctx context.Context, address string) ([]byte, error) {
|
||||
voucherData, err := lookup.Api.VoucherData(ctx, address)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return []byte(voucherData.TokenSymbol), nil
|
||||
}
|
||||
func updateToken(ctx context.Context, store *common.UserDataStore, identity lookup.Identity) error {
|
||||
var api remote.AccountService
|
||||
|
||||
func updateToken(ctx context.Context, store *common.UserDataStore, identity lookup.Identity, tokenAddress string) error {
|
||||
err := updateTokenList(ctx, store, identity)
|
||||
err := updateTokenList(ctx, &api, store, identity)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
store.Db.SetSession(identity.SessionId)
|
||||
activeSym, err := store.ReadEntry(ctx, identity.SessionId, common.DATA_ACTIVE_SYM)
|
||||
activeSym, err := store.ReadEntry(ctx, identity.SessionId, common.DATA_ACTIVE_ADDRESS)
|
||||
if err == nil {
|
||||
return nil
|
||||
}
|
||||
if !db.IsNotFound(err) {
|
||||
return err
|
||||
}
|
||||
if activeSym == nil {
|
||||
activeSym, err = toSym(ctx, tokenAddress)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
logg.Debugf("barfoo")
|
||||
|
||||
err = updateDefaultToken(ctx, store, identity, string(activeSym))
|
||||
if err != nil {
|
||||
@ -162,7 +154,7 @@ func asTokenTransferEvent(gev *geEvent.Event) (*eventTokenTransfer, bool) {
|
||||
}
|
||||
ev.TxHash, err = common.NormalizeHex(gev.TxHash)
|
||||
if err != nil {
|
||||
logg.Errorf("could not decode tx hash", "tx", gev.TxHash, "err", err)
|
||||
logg.Error("could not decode tx hash", "tx", gev.TxHash, "err", err)
|
||||
return nil, false
|
||||
}
|
||||
|
||||
@ -172,12 +164,9 @@ func asTokenTransferEvent(gev *geEvent.Event) (*eventTokenTransfer, bool) {
|
||||
}
|
||||
ev.Value, err = strconv.Atoi(value)
|
||||
if err != nil {
|
||||
logg.Errorf("could not decode value", "value", value, "err", err)
|
||||
logg.Error("could not decode value", "value", value, "err", err)
|
||||
return nil, false
|
||||
}
|
||||
|
||||
ev.VoucherAddress = gev.ContractAddress
|
||||
|
||||
return &ev, true
|
||||
}
|
||||
|
||||
@ -188,7 +177,7 @@ func handleTokenTransfer(ctx context.Context, store *common.UserDataStore, ev *e
|
||||
return err
|
||||
}
|
||||
} else {
|
||||
err = updateToken(ctx, store, identity, ev.VoucherAddress)
|
||||
err = updateToken(ctx, store, identity)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@ -199,7 +188,7 @@ func handleTokenTransfer(ctx context.Context, store *common.UserDataStore, ev *e
|
||||
return err
|
||||
}
|
||||
} else {
|
||||
err = updateToken(ctx, store, identity, ev.VoucherAddress)
|
||||
err = updateToken(ctx, store, identity)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
4
go.mod
4
go.mod
@ -4,9 +4,8 @@ go 1.23.2
|
||||
|
||||
require (
|
||||
git.defalsify.org/vise.git v0.2.1-0.20241031204035-b588301738ed
|
||||
git.grassecon.net/urdt/ussd v0.0.0-20241103143426-0506a8c452f1
|
||||
git.grassecon.net/urdt/ussd v0.0.0-20241102234108-dae12ac4989e
|
||||
github.com/grassrootseconomics/eth-tracker v1.3.0-rc
|
||||
github.com/grassrootseconomics/ussd-data-service v0.0.0-20241003123429-4904b4438a3a
|
||||
github.com/nats-io/nats.go v1.37.0
|
||||
)
|
||||
|
||||
@ -14,6 +13,7 @@ require (
|
||||
github.com/barbashov/iso639-3 v0.0.0-20211020172741-1f4ffb2d8d1c // indirect
|
||||
github.com/fxamacker/cbor/v2 v2.4.0 // indirect
|
||||
github.com/grassrootseconomics/eth-custodial v1.3.0-beta // indirect
|
||||
github.com/grassrootseconomics/ussd-data-service v0.0.0-20241003123429-4904b4438a3a // indirect
|
||||
github.com/graygnuorg/go-gdbm v0.0.0-20220711140707-71387d66dce4 // indirect
|
||||
github.com/jackc/pgpassfile v1.0.0 // indirect
|
||||
github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
|
||||
|
4
go.sum
4
go.sum
@ -1,7 +1,7 @@
|
||||
git.defalsify.org/vise.git v0.2.1-0.20241031204035-b588301738ed h1:4TrsfbK7NKgsa7KjMPlnV/tjYTkAAXP5PWAZzUfzCdI=
|
||||
git.defalsify.org/vise.git v0.2.1-0.20241031204035-b588301738ed/go.mod h1:jyBMe1qTYUz3mmuoC9JQ/TvFeW0vTanCUcPu3H8p4Ck=
|
||||
git.grassecon.net/urdt/ussd v0.0.0-20241103143426-0506a8c452f1 h1:uTscFuyKCqWshcN+pgoJiE0jIVzRrUrgBfI/RsiM7qE=
|
||||
git.grassecon.net/urdt/ussd v0.0.0-20241103143426-0506a8c452f1/go.mod h1:ADB/wpwvI6umvYzGqpJGm/GYj8msxYGiczzWCCdXegs=
|
||||
git.grassecon.net/urdt/ussd v0.0.0-20241102234108-dae12ac4989e h1:YOEAjTrlyDRm0E63fEq8HbMp7oUCyc3vbAPOIw0xNWA=
|
||||
git.grassecon.net/urdt/ussd v0.0.0-20241102234108-dae12ac4989e/go.mod h1:ADB/wpwvI6umvYzGqpJGm/GYj8msxYGiczzWCCdXegs=
|
||||
github.com/alecthomas/assert/v2 v2.2.2 h1:Z/iVC0xZfWTaFNE6bA3z07T86hd45Xe2eLt6WVy2bbk=
|
||||
github.com/alecthomas/assert/v2 v2.2.2/go.mod h1:pXcQ2Asjp247dahGEmsZ6ru0UVwnkhktn7S0bBDLxvQ=
|
||||
github.com/alecthomas/repr v0.2.0 h1:HAzS41CIzNW5syS8Mf9UwXhNH1J9aix/BvDRf1Ml2Yk=
|
||||
|
13
lookup/db.go
13
lookup/db.go
@ -3,15 +3,9 @@ package lookup
|
||||
import (
|
||||
"context"
|
||||
|
||||
"git.defalsify.org/vise.git/db"
|
||||
"git.defalsify.org/vise.git/logging"
|
||||
"git.grassecon.net/urdt/ussd/common"
|
||||
)
|
||||
|
||||
var (
|
||||
logg = logging.NewVanilla().WithDomain("term-lookup")
|
||||
)
|
||||
|
||||
type Identity struct {
|
||||
NormalAddress string
|
||||
ChecksumAddress string
|
||||
@ -35,11 +29,8 @@ func IdentityFromAddress(ctx context.Context, store *common.UserDataStore, addre
|
||||
}
|
||||
|
||||
func getSessionIdByAddress(ctx context.Context, store *common.UserDataStore, address string) (string, error) {
|
||||
// TODO: replace with userdatastore when double sessionid issue fixed
|
||||
//r, err := store.ReadEntry(ctx, address, common.DATA_PUBLIC_KEY_REVERSE)
|
||||
store.Db.SetPrefix(db.DATATYPE_USERDATA)
|
||||
store.Db.SetSession(address)
|
||||
r, err := store.Db.Get(ctx, common.PackKey(common.DATA_PUBLIC_KEY_REVERSE, []byte{}))
|
||||
|
||||
r, err := store.ReadEntry(ctx, address, common.DATA_PUBLIC_KEY_REVERSE)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
@ -1,9 +0,0 @@
|
||||
package lookup
|
||||
|
||||
import (
|
||||
"git.grassecon.net/urdt/ussd/remote"
|
||||
)
|
||||
|
||||
var (
|
||||
Api remote.AccountServiceInterface = &remote.AccountService{}
|
||||
)
|
Loading…
Reference in New Issue
Block a user