Add db type set option

.env.example

@@ -1,7 +1,3 @@
-#Serve Http
-PORT=7123
-HOST=127.0.0.1
-
 #PostgreSQL
 DB_HOST=localhost
 DB_USER=postgres
@@ -19,3 +15,7 @@ DB_TIMEZONE=Africa/Nairobi
 #BALANCE_URL=/api/account/status/
 #CUSTODIAL_URL_BASE=http://localhost:5003
 #DATA_URL_BASE=http://localhost:5006
+
+#Data stream
+#NATS_JETSTREAM_URL=http://localhost:4222
+#NATS_JETSTREAM_CLIENT_NAME=omnom

cmd/main.go

@@ -2,15 +2,15 @@ package main
 
 import (
 	"context"
+	"flag"
 	"fmt"
 	"os"
 	"os/signal"
 	"syscall"
 
-	"git.defalsify.org/vise.git/db/mem"
-
-	"git.grassecon.net/urdt/ussd/config"
 	"git.grassecon.net/urdt/ussd/initializers"
+	"git.grassecon.net/urdt/ussd/common"
+	"git.grassecon.net/term/config"
 	"git.grassecon.net/term/event/nats"
 )
 
@@ -21,15 +21,31 @@ func init() {
 func main() {
 	config.LoadConfig()
 
+	var database string
+	var dbDir string
+
+	flag.StringVar(&database, "db", "gdbm", "database to be used")
+	flag.StringVar(&dbDir, "dbdir", ".state", "database dir to read from")
+	flag.Parse()
+
 	ctx := context.Background()
-	db := mem.NewMemDb()
+	ctx = context.WithValue(ctx, "Database", database)
-	err := db.Connect(ctx, "")
+//	db := mem.NewMemDb()
+//	err := db.Connect(ctx, "")
+//	if err != nil {
+//		fmt.Fprintf(os.Stderr, "Db connect err: %v", err)
+//		os.Exit(1)
+//	}
+
+	menuStorageService := common.NewStorageService(dbDir)
+	err := menuStorageService.EnsureDbDir()
 	if err != nil {
-		fmt.Fprintf(os.Stderr, "Db connect err: %v", err)
+		fmt.Fprintf(os.Stderr, err.Error())
 		os.Exit(1)
 	}
-	n := nats.NewNatsSubscription(db)
+
-	err = n.Connect(ctx, "localhost:4222")
+	n := nats.NewNatsSubscription(menuStorageService)
+	err = n.Connect(ctx, config.JetstreamURL)
 	if err != nil {
 		fmt.Fprintf(os.Stderr, "Stream connect err: %v", err)
 		os.Exit(1)

common/hex.go

@@ -1,18 +0,0 @@
-package common
-
-import (
-	"encoding/hex"
-)
-
-func NormalizeHex(s string) (string, error) {
-	if len(s) >= 2 {
-		if s[:2] == "0x" {
-			s = s[2:]
-		}
-	}
-	r, err := hex.DecodeString(s)
-	if err != nil {
-		return "", err
-	}
-	return hex.EncodeToString(r), nil
-}

config/config.go

@@ -0,0 +1,19 @@
+package config
+
+import (
+	urdtconfig "git.grassecon.net/urdt/ussd/config"
+	"git.grassecon.net/urdt/ussd/initializers"
+)
+
+var (
+	JetstreamURL string
+	JetstreamClientName string
+)
+
+
+func LoadConfig() {
+	urdtconfig.LoadConfig()
+
+	JetstreamURL = initializers.GetEnv("NATS_JETSTREAM_URL", "localhost:4222")
+	JetstreamClientName = initializers.GetEnv("NATS_JETSTREAM_CLIENT_NAME", "omnom")
+}

event/custodial_registration.go

@@ -5,18 +5,22 @@ import (
 
 	geEvent "github.com/grassrootseconomics/eth-tracker/pkg/event"
 
+	"git.defalsify.org/vise.git/persist"
 	"git.grassecon.net/urdt/ussd/common"
 	"git.grassecon.net/term/lookup"
 )
 
 const (
 	evReg = "CUSTODIAL_REGISTRATION"
+	accountCreatedFlag = 9
 )
 
+// fields used for handling custodial registration event.
 type eventCustodialRegistration struct {
 	Account string
 }
 
+// attempt to coerce event as custodial registration.
 func asCustodialRegistrationEvent(gev *geEvent.Event) (*eventCustodialRegistration, bool) {
 	var ok bool
 	var ev eventCustodialRegistration
@@ -32,11 +36,21 @@ func asCustodialRegistrationEvent(gev *geEvent.Event) (*eventCustodialRegistrati
 	return &ev, true
 }
 
-func handleCustodialRegistration(ctx context.Context, store *common.UserDataStore, ev *eventCustodialRegistration) error {
+// handle custodial registration.
+//
+// TODO: implement account created in userstore instead, so that
+// the need for persister and state use here is eliminated (it
+// introduces concurrency risks)
+func handleCustodialRegistration(ctx context.Context, store *common.UserDataStore, pr *persist.Persister, ev *eventCustodialRegistration) error {
 	identity, err := lookup.IdentityFromAddress(ctx, store, ev.Account)
 	if err != nil {
 		return err
 	}
-	_ = identity
+	err = pr.Load(identity.SessionId)
-	return nil
+	if err != nil {
+		return err
+	}
+	st := pr.GetState()
+	st.SetFlag(accountCreatedFlag)
+	return pr.Save(identity.SessionId)
 }

event/custodial_test.go

@@ -0,0 +1,62 @@
+package event
+
+import (
+	"context"
+	"testing"
+
+	memdb "git.defalsify.org/vise.git/db/mem"
+	"git.defalsify.org/vise.git/db"
+	"git.defalsify.org/vise.git/persist"
+	"git.defalsify.org/vise.git/state"
+	"git.defalsify.org/vise.git/cache"
+	"git.grassecon.net/urdt/ussd/config"
+	"git.grassecon.net/urdt/ussd/common"
+	"git.grassecon.net/term/internal/testutil"
+)
+
+func TestCustodialRegistration(t *testing.T) {
+	err := config.LoadConfig()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	ctx := context.Background()
+	userDb := memdb.NewMemDb()
+	err = userDb.Connect(ctx, "")
+	if err != nil {
+		panic(err)
+	}
+
+	alice, err := common.NormalizeHex(testutil.AliceChecksum)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	userDb.SetSession(alice)
+	userDb.SetPrefix(db.DATATYPE_USERDATA)
+	err = userDb.Put(ctx, common.PackKey(common.DATA_PUBLIC_KEY_REVERSE, []byte{}), []byte(testutil.AliceSession))
+	if err != nil {
+		t.Fatal(err)
+	}
+	store := common.UserDataStore{
+		Db: userDb,
+	}
+
+	st := state.NewState(248)
+	ca := cache.NewCache()
+	pr := persist.NewPersister(userDb)
+	pr = pr.WithContent(st, ca)
+	err = pr.Save(testutil.AliceSession)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	ev := &eventCustodialRegistration{
+		Account: testutil.AliceChecksum,
+	}
+	err = handleCustodialRegistration(ctx, &store, pr, ev)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+}

event/nats/nats.go

@@ -3,20 +3,24 @@ package nats
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 
 	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"
+	"git.grassecon.net/term/config"
 )
 
 var (
 	logg = logging.NewVanilla().WithDomain("term-nats")
 )
 
+// NatsSubscription encapsulates the jetstream session providing events.
+//
+// Extends Router.
 type NatsSubscription struct {
 	event.Router
 	ctx context.Context
@@ -26,16 +30,21 @@ type NatsSubscription struct {
 	cctx jetstream.ConsumeContext
 }
 
-func NewNatsSubscription(store db.Db) *NatsSubscription {
+// NewNatsSubscription creates a new NatsSubscription with the given user store.
+func NewNatsSubscription(store common.StorageServices) *NatsSubscription {
 	return &NatsSubscription{
 		Router: event.Router{
-			Store: &common.UserDataStore{
+			Store: store,
-				Db: store,
-			},
 		},
 	}
 }
 
+// Connect sets up the connection to the nats server and a consumer for the 
+// "Jetstream".
+//
+// Fails if connection fails or the "Jetstream" consumer cannot be set up.
+//
+// Once connected, it will attempt to reconnect if disconnected.
 func(n *NatsSubscription) Connect(ctx context.Context, connStr string) error {
 	var err error
 
@@ -43,19 +52,23 @@ func(n *NatsSubscription) Connect(ctx context.Context, connStr string) error {
 	if err != nil {
 		return err
 	}
+	n.conn.SetDisconnectErrHandler(disconnectHandler)
+	n.conn.SetReconnectHandler(reconnectHandler)
 	n.js, err = jetstream.New(n.conn)
 	if err != nil {
 		return err
 	}
 	n.cs, err = n.js.CreateConsumer(ctx, "TRACKER", jetstream.ConsumerConfig{
-		Name: "omnom",
+		Name: config.JetstreamClientName,
-		Durable: "omnom",
+		Durable: config.JetstreamClientName,
 		FilterSubjects: []string{"TRACKER.*"},
 	})
 	if err != nil {
 		return err
 	}
 
+	serverInfo := toServerInfo(n.conn)
+	logg.DebugCtxf(ctx, "nats connected, starting consumer", "status", n.conn.Status(), "server", serverInfo)
 	n.ctx = ctx
 	n.cctx, err = n.cs.Consume(n.handleEvent)
 	if err != nil {
@@ -65,6 +78,7 @@ func(n *NatsSubscription) Connect(ctx context.Context, connStr string) error {
 	return nil
 }
 
+// Close cleanly brings down the nats and jetstream connection.
 func(n *NatsSubscription) Close() error {
 	n.cctx.Stop()
 	select {
@@ -74,13 +88,7 @@ func(n *NatsSubscription) Close() error {
 	return nil
 }
 
-func fail(m jetstream.Msg) {
+// jetstream message handler and acknowledger.
-	err := m.Nak()
-	if err != nil {
-		logg.Errorf("nats nak fail", "err", err)
-	}
-}
-
 func(n *NatsSubscription) handleEvent(m jetstream.Msg) {
 	var ev geEvent.Event
 
@@ -104,3 +112,27 @@ func(n *NatsSubscription) handleEvent(m jetstream.Msg) {
 	}
 	logg.DebugCtxf(n.ctx, "handle msg complete")
 }
+
+// used if message should be retried.
+func fail(m jetstream.Msg) {
+	err := m.Nak()
+	if err != nil {
+		logg.Errorf("nats nak fail", "err", err)
+	}
+}
+
+// server info string for debug.
+func toServerInfo(conn *nats.Conn) string {
+	return fmt.Sprintf("%s@%s (v%s)", conn.ConnectedServerName(), conn.ConnectedUrlRedacted(), conn.ConnectedServerVersion())
+}
+
+// on nats disconnection.
+func disconnectHandler(conn *nats.Conn, err error) {
+	logg.Errorf("nats disconnected", "status", conn.Status(), "reconnects", conn.Stats().Reconnects, "err", err)
+}
+
+// on nats reconnection.
+func reconnectHandler(conn *nats.Conn) {
+	serverInfo := toServerInfo(conn)
+	logg.Errorf("nats reconnected", "status", conn.Status(), "reconnects", conn.Stats().Reconnects, "server", serverInfo)
+}

event/nats/nats_test.go

@@ -3,6 +3,8 @@ package nats
 import (
 	"bytes"
 	"context"
+	"fmt"
+	"strconv"
 	"testing"
 	"time"
 
@@ -16,13 +18,20 @@ import (
 	"git.grassecon.net/urdt/ussd/models"
 	"git.grassecon.net/term/lookup"
 	"git.grassecon.net/term/event"
+	"git.grassecon.net/term/internal/testutil"
 )
 
-func init() {
-}
-
 const (
-	aliceChecksum = "0xeae046BF396e91f5A8D74f863dC57c107c8a4a70"
+	txBlock = 42
+	tokenAddress = "0x765DE816845861e75A25fCA122bb6898B8B1282a"
+	tokenSymbol = "FOO"
+	tokenName = "Foo Token"
+	tokenDecimals = 6
+	txValue = 1337
+	tokenBalance = 362436
+	txTimestamp = 1730592500
+	txHash = "0xabcdef0123456789abcdef0123456789abcdef0123456789abcdef0123456789"
+	sinkAddress = "0xb42C5920014eE152F2225285219407938469BBfA"
 )
 
 // TODO: jetstream, would have been nice of you to provide an easier way to make a mock msg
@@ -78,65 +87,40 @@ 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) {
-	logg.DebugCtxf(ctx, "mockapi fetchtransactions", "key", publicKey)
-	return []dataserviceapi.Last10TxResponse{
-		dataserviceapi.Last10TxResponse{
-			Sender: "0xeae046BF396e91f5A8D74f863dC57c107c8a4a70",
-			Recipient: "B3117202371853e24B725d4169D87616A7dDb127",
-			TransferValue: "1337",
-			ContractAddress: "0x765DE816845861e75A25fCA122bb6898B8B1282a",
-			TxHash: "0xabcdef0123456789abcdef0123456789abcdef0123456789abcdef0123456789",
-			DateBlock: time.Unix(1730592500, 0),
-			TokenSymbol: "FOO",
-			TokenDecimals: "6",
-		},
-	}, 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{}
+	api := &testutil.MockApi{}
+	api.TransactionsContent = []dataserviceapi.Last10TxResponse{
+		dataserviceapi.Last10TxResponse{
+			Sender: testutil.AliceChecksum,
+			Recipient: testutil.BobChecksum,
+			TransferValue: strconv.Itoa(txValue),
+			ContractAddress: tokenAddress,
+			TxHash: txHash,
+			DateBlock: time.Unix(txTimestamp, 0),
+			TokenSymbol: tokenSymbol,
+			TokenDecimals: strconv.Itoa(tokenDecimals),
+		},
+	}
+	api.VoucherDataContent = &models.VoucherDataResult{
+		TokenSymbol: tokenSymbol,
+		TokenName: tokenName,
+		TokenDecimals: strconv.Itoa(tokenDecimals),
+		SinkAddress: sinkAddress,
+	}
+	api.VouchersContent = []dataserviceapi.TokenHoldings{
+		dataserviceapi.TokenHoldings{
+			ContractAddress: tokenAddress,
+			TokenSymbol: tokenSymbol,
+			TokenDecimals: strconv.Itoa(tokenDecimals),
+			Balance: strconv.Itoa(tokenBalance),
+		},
+	}
+	lookup.Api = api
 
 	ctx := context.Background()
 	userDb := memdb.NewMemDb()
@@ -145,34 +129,36 @@ func TestHandleMsg(t *testing.T) {
 		panic(err)
 	}
 
-	alice, err := common.NormalizeHex(aliceChecksum)
+	alice, err := common.NormalizeHex(testutil.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))
+	err = userDb.Put(ctx, common.PackKey(common.DATA_PUBLIC_KEY_REVERSE, []byte{}), []byte(testutil.AliceSession))
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	sub := NewNatsSubscription(userDb)
+	storageService := &testutil.TestStorageService{
+		Store: userDb,
+	}
+	sub := NewNatsSubscription(storageService)
 
-	data := `{
+	data := fmt.Sprintf(`{
-	"block": 42,
+	"block": %d,
-	"contractAddress": "0x765DE816845861e75A25fCA122bb6898B8B1282a",
+	"contractAddress": "%s",
 	"success": true,
-	"timestamp": 1730592500,
+	"timestamp": %d,
-	"transactionHash": "0xabcdef0123456789abcdef0123456789abcdef0123456789abcdef0123456789",
+	"transactionHash": "%s",
 	"transactionType": "TOKEN_TRANSFER",
 	"payload": {
-		"from": "0xeae046BF396e91f5A8D74f863dC57c107c8a4a70",
+		"from": "%s",
-		"to": "B3117202371853e24B725d4169D87616A7dDb127",
+		"to": "%s",
-		"value": "1337"
+		"value": "%d"
 	}
-}`
+}`, txBlock, tokenAddress, txTimestamp, txHash, testutil.AliceChecksum, testutil.BobChecksum, txValue)
 	msg := &testMsg{
 		data: []byte(data),
 	}
@@ -181,23 +167,23 @@ func TestHandleMsg(t *testing.T) {
 	store := common.UserDataStore{
 		Db: userDb,
 	}
-	v, err := store.ReadEntry(ctx, aliceSession, common.DATA_ACTIVE_SYM)
+	v, err := store.ReadEntry(ctx, testutil.AliceSession, common.DATA_ACTIVE_SYM)
 	if err != nil {
 		t.Fatal(err)
 	}
-	if !bytes.Equal(v, []byte("FOO")) {
+	if !bytes.Equal(v, []byte(tokenSymbol)) {
-		t.Fatalf("expected 'FOO', got %s", v)
+		t.Fatalf("expected '%s', got %s", tokenSymbol, v)
 	}
 
-	v, err = store.ReadEntry(ctx, aliceSession, common.DATA_ACTIVE_BAL)
+	v, err = store.ReadEntry(ctx, testutil.AliceSession, common.DATA_ACTIVE_BAL)
 	if err != nil {
 		t.Fatal(err)
 	}
-	if !bytes.Equal(v, []byte("362436")) {
+	if !bytes.Equal(v, []byte(strconv.Itoa(tokenBalance))) {
-		t.Fatalf("expected '362436', got %s", v)
+		t.Fatalf("expected '%d', got %s", tokenBalance, v)
 	}
 
-	v, err = store.ReadEntry(ctx, aliceSession, common.DATA_TRANSACTIONS)
+	v, err = store.ReadEntry(ctx, testutil.AliceSession, common.DATA_TRANSACTIONS)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -206,13 +192,13 @@ func TestHandleMsg(t *testing.T) {
 	}
 
 	userDb.SetPrefix(event.DATATYPE_USERSUB)
-	userDb.SetSession(aliceSession)
+	userDb.SetSession(testutil.AliceSession)
 	k := append([]byte("vouchers"), []byte("sym")...)
 	v, err = userDb.Get(ctx, k)
 	if err != nil {
 		t.Fatal(err)
 	}
-	if !bytes.Contains(v, []byte("1:FOO")) {
+	if !bytes.Contains(v, []byte(fmt.Sprintf("1:%s", tokenSymbol))) {
-		t.Fatalf("expected '1:FOO', got %s", v)
+		t.Fatalf("expected '1:%s', got %s", tokenSymbol, v)
 	}
 }

event/route.go

@@ -10,28 +10,40 @@ import (
 	"git.grassecon.net/urdt/ussd/common"
 )
 
-// TODO: this vocabulary should be public in and provided by the eth-tracker repo
-const (
-	evGive = "FAUCET_GIVE"
-)
-
 var (
 	logg = logging.NewVanilla().WithDomain("term-event")
 )
 
+// Router is responsible for invoking handlers corresponding to events.
 type Router struct {
-	Store *common.UserDataStore
+	Store common.StorageServices
 }
 
+// Route parses an event from the event stream, and resolves the handler
+// corresponding to the event.
+//
+// An error will be returned if no handler can be found, or if the resolved
+// handler fails to successfully execute.
 func(r *Router) Route(ctx context.Context, gev *geEvent.Event) error {
 	logg.DebugCtxf(ctx, "have event", "ev", gev)
+	store, err := r.Store.GetUserdataDb(ctx)
+	if err != nil {
+		return err
+	}
+	userStore := &common.UserDataStore{
+		Db: store,
+	}
 	evCC, ok := asCustodialRegistrationEvent(gev)
 	if ok {
-		return handleCustodialRegistration(ctx, r.Store, evCC)
+		pr, err := r.Store.GetPersister(ctx)
+		if err != nil {
+			return err
+		}
+		return handleCustodialRegistration(ctx, userStore, pr, evCC)
 	}
 	evTT, ok := asTokenTransferEvent(gev)
 	if ok {
-		return handleTokenTransfer(ctx, r.Store, evTT)
+		return handleTokenTransfer(ctx, userStore, evTT)
 	}
 
 	return fmt.Errorf("unexpected message")

event/sub.go

@@ -1,12 +0,0 @@
-package event
-
-import (
-	"context"
-	"io"
-)
-
-type Subscription interface {
-	io.Closer
-	Connect(ctx context.Context, connStr string) error
-	Next() error
-}

event/token.go

@@ -16,26 +16,35 @@ import (
 
 const (
 	evTokenTransfer = "TOKEN_TRANSFER"
-	// TODO: use export from urdt storage
+	// TODO: export from urdt storage package
 	DATATYPE_USERSUB = 64
 )
 
-func renderTx() {
+// fields used for handling token transfer event.
-
+type eventTokenTransfer struct {
+	To string
+	Value int
+	VoucherAddress string
+	TxHash string
+	From string
 }
 
-type eventTokenTransfer struct {
+type eventTokenMint struct {
-	From string
 	To string
 	Value int
 	TxHash string
 	VoucherAddress string
 }
 
+
+// formatter for transaction data
+//
+// TODO: current formatting is a placeholder.
 func formatTransaction(idx int, tx dataserviceapi.Last10TxResponse) string {
 	return fmt.Sprintf("%d %s %s", idx, tx.DateBlock, tx.TxHash[:10])
 }
 
+// refresh and store transaction history.
 func updateTokenTransferList(ctx context.Context, store *common.UserDataStore, identity lookup.Identity) error {
 	var r []string
 
@@ -53,6 +62,9 @@ func updateTokenTransferList(ctx context.Context, store *common.UserDataStore, i
 	return store.WriteEntry(ctx, identity.SessionId, common.DATA_TRANSACTIONS, []byte(s))
 }
 
+// refresh and store token list.
+//
+// TODO: when subprefixdb has been exported, can use function in ...urdt/ussd/common/ instead
 func updateTokenList(ctx context.Context, store *common.UserDataStore, identity lookup.Identity) error {
 	holdings, err := lookup.Api.FetchVouchers(ctx, identity.ChecksumAddress)
 	if err != nil {
@@ -61,7 +73,6 @@ func updateTokenList(ctx context.Context, store *common.UserDataStore, identity
 	metadata := common.ProcessVouchers(holdings)
 	_ = metadata
 
-	// 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)
@@ -94,6 +105,7 @@ func updateTokenList(ctx context.Context, store *common.UserDataStore, identity
 	return nil
 }
 
+// set default token to given symbol.
 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?
@@ -101,14 +113,15 @@ func updateDefaultToken(ctx context.Context, store *common.UserDataStore, identi
 	if err != nil {
 		return err
 	}
-	logg.TraceCtxf(ctx, "tokendaa", "d", tokenData)
 	return common.UpdateVoucherData(ctx, store, identity.SessionId, tokenData)
 }
 
+// waiter to check whether object is available on dependency endpoints.
 func updateWait(ctx context.Context) error {
 	return nil
 }
 
+// use api to resolve address to token symbol.
 func toSym(ctx context.Context, address string) ([]byte, error) {
 	voucherData, err := lookup.Api.VoucherData(ctx, address)
 	if err != nil {
@@ -117,6 +130,7 @@ func toSym(ctx context.Context, address string) ([]byte, error) {
 	return []byte(voucherData.TokenSymbol), nil
 }
 
+// execute all 
 func updateToken(ctx context.Context, store *common.UserDataStore, identity lookup.Identity, tokenAddress string) error {
 	err := updateTokenList(ctx, store, identity)
 	if err != nil {
@@ -152,6 +166,7 @@ func updateToken(ctx context.Context, store *common.UserDataStore, identity look
 	return nil
 }
 
+// attempt to coerce event as token transfer event.
 func asTokenTransferEvent(gev *geEvent.Event) (*eventTokenTransfer, bool) {
 	var err error
 	var ok bool
@@ -162,7 +177,7 @@ func asTokenTransferEvent(gev *geEvent.Event) (*eventTokenTransfer, bool) {
 	}
 
 	pl := gev.Payload
-	// assuming from and to are checksum addresses
+	// we are assuming from and to are checksum addresses
 	ev.From, ok = pl["from"].(string)
 	if !ok {
 		return nil, false
@@ -192,6 +207,9 @@ func asTokenTransferEvent(gev *geEvent.Event) (*eventTokenTransfer, bool) {
 	return &ev, true
 }
 
+// handle token transfer.
+//
+// if from and to are NOT the same, handle code will be executed once for each side of the transfer.
 func handleTokenTransfer(ctx context.Context, store *common.UserDataStore, ev *eventTokenTransfer) error {
 	identity, err := lookup.IdentityFromAddress(ctx, store, ev.From)
 	if err != nil {
@@ -204,7 +222,27 @@ func handleTokenTransfer(ctx context.Context, store *common.UserDataStore, ev *e
 			return err
 		}
 	}
-	identity, err = lookup.IdentityFromAddress(ctx, store, ev.To)
+
+	if strings.Compare(ev.To, ev.From) != 0 {
+		identity, err = lookup.IdentityFromAddress(ctx, store, ev.To)
+		if err != nil {
+			if !db.IsNotFound(err) {
+				return err
+			}
+		} else {
+			err = updateToken(ctx, store, identity, ev.VoucherAddress)
+			if err != nil {
+				return err
+			}
+		}
+	}
+
+	return nil
+}
+
+// handle token mint.
+func handleTokenMint(ctx context.Context, store *common.UserDataStore, ev *eventTokenMint) error {
+	identity, err := lookup.IdentityFromAddress(ctx, store, ev.To)
 	if err != nil {
 		if !db.IsNotFound(err) {
 			return err
@@ -215,6 +253,5 @@ func handleTokenTransfer(ctx context.Context, store *common.UserDataStore, ev *e
 			return err
 		}
 	}
-
 	return nil
 }

event/token_test.go

@@ -0,0 +1,241 @@
+package event
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"strconv"
+	"testing"
+	"time"
+
+	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/config"
+	"git.grassecon.net/urdt/ussd/models"
+	"git.grassecon.net/term/lookup"
+	"git.grassecon.net/urdt/ussd/common"
+	"git.grassecon.net/term/internal/testutil"
+)
+
+const (
+	txBlock = 42
+	tokenAddress = "0x765DE816845861e75A25fCA122bb6898B8B1282a"
+	tokenSymbol = "FOO"
+	tokenName = "Foo Token"
+	tokenDecimals = 6
+	txValue = 1337
+	tokenBalance = 362436
+	txTimestamp = 1730592500
+	txHash = "0xabcdef0123456789abcdef0123456789abcdef0123456789abcdef0123456789"
+	sinkAddress = "0xb42C5920014eE152F2225285219407938469BBfA"
+)
+
+func TestTokenTransfer(t *testing.T) {
+	err := config.LoadConfig()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	api := &testutil.MockApi{}
+	api.TransactionsContent = []dataserviceapi.Last10TxResponse{
+		dataserviceapi.Last10TxResponse{
+			Sender: testutil.AliceChecksum,
+			Recipient: testutil.BobChecksum,
+			TransferValue: strconv.Itoa(txValue),
+			ContractAddress: tokenAddress,
+			TxHash: txHash,
+			DateBlock: time.Unix(txTimestamp, 0),
+			TokenSymbol: tokenSymbol,
+			TokenDecimals: strconv.Itoa(tokenDecimals),
+		},
+	}
+	api.VoucherDataContent = &models.VoucherDataResult{
+		TokenSymbol: tokenSymbol,
+		TokenName: tokenName,
+		TokenDecimals: strconv.Itoa(tokenDecimals),
+		SinkAddress: sinkAddress,
+	}
+	api.VouchersContent = []dataserviceapi.TokenHoldings{
+		dataserviceapi.TokenHoldings{
+			ContractAddress: tokenAddress,
+			TokenSymbol: tokenSymbol,
+			TokenDecimals: strconv.Itoa(tokenDecimals),
+			Balance: strconv.Itoa(tokenBalance),
+		},
+	}
+	lookup.Api = api
+
+	ctx := context.Background()
+	userDb := memdb.NewMemDb()
+	err = userDb.Connect(ctx, "")
+	if err != nil {
+		panic(err)
+	}
+
+	alice, err := common.NormalizeHex(testutil.AliceChecksum)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// TODO: deduplicate test setup
+	userDb.SetSession(alice)
+	userDb.SetPrefix(db.DATATYPE_USERDATA)
+	err = userDb.Put(ctx, common.PackKey(common.DATA_PUBLIC_KEY_REVERSE, []byte{}), []byte(testutil.AliceSession))
+	if err != nil {
+		t.Fatal(err)
+	}
+	store := common.UserDataStore{
+		Db: userDb,
+	}
+
+	ev := &eventTokenTransfer{
+		From: testutil.BobChecksum,
+		To: testutil.AliceChecksum,
+		Value: txValue,
+	}
+	err = handleTokenTransfer(ctx, &store, ev)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	v, err := store.ReadEntry(ctx, testutil.AliceSession, common.DATA_ACTIVE_SYM)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !bytes.Equal(v, []byte(tokenSymbol)) {
+		t.Fatalf("expected '%s', got %s", tokenSymbol, v)
+	}
+
+	v, err = store.ReadEntry(ctx, testutil.AliceSession, common.DATA_ACTIVE_BAL)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !bytes.Equal(v, []byte(strconv.Itoa(tokenBalance))) {
+		t.Fatalf("expected '%d', got %s", tokenBalance, v)
+	}
+
+	v, err = store.ReadEntry(ctx, testutil.AliceSession, common.DATA_TRANSACTIONS)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !bytes.Contains(v, []byte("abcdef")) {
+		t.Fatal("no transaction data")
+	}
+
+	userDb.SetPrefix(DATATYPE_USERSUB)
+	userDb.SetSession(testutil.AliceSession)
+	k := append([]byte("vouchers"), []byte("sym")...)
+	v, err = userDb.Get(ctx, k)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !bytes.Contains(v, []byte(fmt.Sprintf("1:%s", tokenSymbol))) {
+		t.Fatalf("expected '1:%s', got %s", tokenSymbol, v)
+	}
+
+
+}
+
+func TestTokenMint(t *testing.T) {
+	err := config.LoadConfig()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	api := &testutil.MockApi{}
+	api.TransactionsContent = []dataserviceapi.Last10TxResponse{
+		dataserviceapi.Last10TxResponse{
+			Sender: testutil.AliceChecksum,
+			Recipient: testutil.BobChecksum,
+			TransferValue: strconv.Itoa(txValue),
+			ContractAddress: tokenAddress,
+			TxHash: txHash,
+			DateBlock: time.Unix(txTimestamp, 0),
+			TokenSymbol: tokenSymbol,
+			TokenDecimals: strconv.Itoa(tokenDecimals),
+		},
+	}
+	api.VoucherDataContent = &models.VoucherDataResult{
+		TokenSymbol: tokenSymbol,
+		TokenName: tokenName,
+		TokenDecimals: strconv.Itoa(tokenDecimals),
+		SinkAddress: sinkAddress,
+	}
+	api.VouchersContent = []dataserviceapi.TokenHoldings{
+		dataserviceapi.TokenHoldings{
+			ContractAddress: tokenAddress,
+			TokenSymbol: tokenSymbol,
+			TokenDecimals: strconv.Itoa(tokenDecimals),
+			Balance: strconv.Itoa(tokenBalance),
+		},
+	}
+	lookup.Api = api
+
+	ctx := context.Background()
+	userDb := memdb.NewMemDb()
+	err = userDb.Connect(ctx, "")
+	if err != nil {
+		panic(err)
+	}
+
+	alice, err := common.NormalizeHex(testutil.AliceChecksum)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	userDb.SetSession(alice)
+	userDb.SetPrefix(db.DATATYPE_USERDATA)
+	err = userDb.Put(ctx, common.PackKey(common.DATA_PUBLIC_KEY_REVERSE, []byte{}), []byte(testutil.AliceSession))
+	if err != nil {
+		t.Fatal(err)
+	}
+	store := common.UserDataStore{
+		Db: userDb,
+	}
+
+	ev := &eventTokenMint{
+		To: testutil.AliceChecksum,
+		Value: txValue,
+	}
+	err = handleTokenMint(ctx, &store, ev)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	v, err := store.ReadEntry(ctx, testutil.AliceSession, common.DATA_ACTIVE_SYM)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !bytes.Equal(v, []byte(tokenSymbol)) {
+		t.Fatalf("expected '%s', got %s", tokenSymbol, v)
+	}
+
+	v, err = store.ReadEntry(ctx, testutil.AliceSession, common.DATA_ACTIVE_BAL)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !bytes.Equal(v, []byte(strconv.Itoa(tokenBalance))) {
+		t.Fatalf("expected '%d', got %s", tokenBalance, v)
+	}
+
+	v, err = store.ReadEntry(ctx, testutil.AliceSession, common.DATA_TRANSACTIONS)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !bytes.Contains(v, []byte("abcdef")) {
+		t.Fatal("no transaction data")
+	}
+
+	userDb.SetPrefix(DATATYPE_USERSUB)
+	userDb.SetSession(testutil.AliceSession)
+	k := append([]byte("vouchers"), []byte("sym")...)
+	v, err = userDb.Get(ctx, k)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !bytes.Contains(v, []byte(fmt.Sprintf("1:%s", tokenSymbol))) {
+		t.Fatalf("expected '1:%s', got %s", tokenSymbol, v)
+	}
+
+}

go.mod

@@ -4,7 +4,7 @@ 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-20241104231804-859de0513ae5
 	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

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-20241104231804-859de0513ae5 h1:BNj1ym9I3AzMwc5yjmfKueLCCW/2IfrYa7WuvHYDLBo=
-git.grassecon.net/urdt/ussd v0.0.0-20241103143426-0506a8c452f1/go.mod h1:ADB/wpwvI6umvYzGqpJGm/GYj8msxYGiczzWCCdXegs=
+git.grassecon.net/urdt/ussd v0.0.0-20241104231804-859de0513ae5/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=

handler/error.go

@@ -1,9 +0,0 @@
-package handler
-
-import (
-	"fmt"
-)
-
-var (
-	ErrInvalidPayload = fmt.Errorf("Invalid event payload")
-)

internal/testutil/api.go

@@ -0,0 +1,51 @@
+package testutil
+
+import (
+	"context"
+
+	"git.defalsify.org/vise.git/logging"
+	dataserviceapi "github.com/grassrootseconomics/ussd-data-service/pkg/api"
+	"git.grassecon.net/urdt/ussd/models"
+)
+
+var (
+	logg = logging.NewVanilla().WithDomain("term-testutiL")
+)
+
+const (
+	AliceChecksum = "0xeae046BF396e91f5A8D74f863dC57c107c8a4a70"
+	BobChecksum = "0xB3117202371853e24B725d4169D87616A7dDb127"
+	AliceSession = "5553425"
+)
+
+type MockApi struct {
+	TransactionsContent []dataserviceapi.Last10TxResponse
+	VouchersContent []dataserviceapi.TokenHoldings
+	VoucherDataContent *models.VoucherDataResult
+}
+
+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 m.VouchersContent, nil
+}
+
+func(m MockApi) FetchTransactions(ctx context.Context, publicKey string) ([]dataserviceapi.Last10TxResponse, error) {
+	logg.DebugCtxf(ctx, "mockapi fetchtransactions", "key", publicKey)
+	return m.TransactionsContent, nil
+}
+
+func(m MockApi) VoucherData(ctx context.Context, address string) (*models.VoucherDataResult, error) {
+	return m.VoucherDataContent, nil
+}

internal/testutil/service.go

@@ -0,0 +1,35 @@
+package testutil
+
+import (
+	"context"
+	"errors"
+
+	"git.defalsify.org/vise.git/persist"
+	"git.defalsify.org/vise.git/resource"
+	"git.defalsify.org/vise.git/db"
+)
+
+// TestStorageService wraps db for nats subscription.
+type TestStorageService struct {
+	Store  db.Db
+}
+
+// GetUserdataDb implements urdt/ussd/common.StorageServices.
+func(ss *TestStorageService) GetUserdataDb(ctx context.Context) (db.Db, error) {
+	return ss.Store, nil
+}
+
+// GetPersister implements urdt/ussd/common.StorageServices.
+func(ts *TestStorageService) GetPersister(ctx context.Context) (*persist.Persister, error) {
+	return persist.NewPersister(ts.Store), nil
+}
+
+// GetResource implements urdt/ussd/common.StorageServices.
+func(ts *TestStorageService) GetResource(ctx context.Context) (resource.Resource, error) {
+	return nil, errors.New("not implemented")
+}
+
+// EnsureDbDir implements urdt/ussd/common.StorageServices.
+func(ss *TestStorageService) EnsureDbDir() error {
+	return nil
+}

lookup/db.go

@@ -12,12 +12,19 @@ var (
 	logg = logging.NewVanilla().WithDomain("term-lookup")
 )
 
+// Identity contains all flavors of identifiers used across stream, api and
+// client for a single agent.
 type Identity struct {
 	NormalAddress string
 	ChecksumAddress string
 	SessionId string
 }
 
+// IdentityFromAddress fully populates and Identity object from a given
+// checksum address.
+//
+// It is the caller's responsibility to ensure that a valid checksum address
+// is passed.
 func IdentityFromAddress(ctx context.Context, store *common.UserDataStore, address string) (Identity, error) {
 	var err error
 	var ident Identity
@@ -34,6 +41,7 @@ func IdentityFromAddress(ctx context.Context, store *common.UserDataStore, addre
 	return ident, nil
 }
 
+// load matching session from address from db store.
 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)

lookup/token.go

@@ -5,5 +5,6 @@ import (
 )
 
 var (
+	// Api provides the api implementation for all external lookups.
 	Api remote.AccountServiceInterface = &remote.AccountService{}
 )