Merge pull request 'africas-talking' (#50) from africas-talking into lash/async-driver

Reviewed-on: urdt/ussd#50
Reviewed-by: lash <accounts-grassrootseconomics@holbrook.no>

cmd/africastalking/main.go

@@ -0,0 +1,257 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"net/http"
+	"os"
+	"os/signal"
+	"path"
+	"strconv"
+	"strings"
+	"syscall"
+
+	"git.defalsify.org/vise.git/asm"
+	"git.defalsify.org/vise.git/db"
+	fsdb "git.defalsify.org/vise.git/db/fs"
+	gdbmdb "git.defalsify.org/vise.git/db/gdbm"
+	"git.defalsify.org/vise.git/engine"
+	"git.defalsify.org/vise.git/logging"
+	"git.defalsify.org/vise.git/resource"
+
+	"git.grassecon.net/urdt/ussd/internal/handlers"
+	"git.grassecon.net/urdt/ussd/internal/handlers/ussd"
+	httpserver "git.grassecon.net/urdt/ussd/internal/http"
+)
+
+var (
+	logg = logging.NewVanilla()
+	scriptDir = path.Join("services", "registration")
+)
+
+type atRequestParser struct {}
+
+func(arp *atRequestParser) GetSessionId(rq any) (string, error) {
+	rqv, ok := rq.(*http.Request)
+	if !ok {
+		return "", handlers.ErrInvalidRequest
+	}
+	if err := rqv.ParseForm(); err != nil {
+		return "", fmt.Errorf("failed to parse form data: %v", err)
+	}
+
+	phoneNumber := rqv.FormValue("phoneNumber")
+	if phoneNumber == "" {
+		return "", fmt.Errorf("no phone number found")
+	}
+
+	return phoneNumber, nil
+}
+
+func(arp *atRequestParser) GetInput(rq any) ([]byte, error) {
+	rqv, ok := rq.(*http.Request)
+	if !ok {
+		return nil, handlers.ErrInvalidRequest
+	}
+	if err := rqv.ParseForm(); err != nil {
+		return nil, fmt.Errorf("failed to parse form data: %v", err)
+	}
+
+	text := rqv.FormValue("text")
+
+	parts := strings.Split(text, "*")
+	if len(parts) == 0 {
+		return nil, fmt.Errorf("no input found")
+	}
+
+	return []byte(parts[len(parts)-1]), nil
+}
+
+
+func getFlags(fp string, debug bool) (*asm.FlagParser, error) {
+	flagParser := asm.NewFlagParser().WithDebug()
+	_, err := flagParser.Load(fp)
+	if err != nil {
+		return nil, err
+	}
+	return flagParser, nil
+}
+
+func getHandler(appFlags *asm.FlagParser, rs *resource.DbResource, userdataStore db.Db) (*ussd.Handlers, error) {
+
+	ussdHandlers, err := ussd.NewHandlers(appFlags, userdataStore)
+	if err != nil {
+		return nil, err
+	}
+	rs.AddLocalFunc("select_language", ussdHandlers.SetLanguage)
+	rs.AddLocalFunc("create_account", ussdHandlers.CreateAccount)
+	rs.AddLocalFunc("save_pin", ussdHandlers.SavePin)
+	rs.AddLocalFunc("verify_pin", ussdHandlers.VerifyPin)
+	rs.AddLocalFunc("check_identifier", ussdHandlers.CheckIdentifier)
+	rs.AddLocalFunc("check_account_status", ussdHandlers.CheckAccountStatus)
+	rs.AddLocalFunc("authorize_account", ussdHandlers.Authorize)
+	rs.AddLocalFunc("quit", ussdHandlers.Quit)
+	rs.AddLocalFunc("check_balance", ussdHandlers.CheckBalance)
+	rs.AddLocalFunc("validate_recipient", ussdHandlers.ValidateRecipient)
+	rs.AddLocalFunc("transaction_reset", ussdHandlers.TransactionReset)
+	rs.AddLocalFunc("max_amount", ussdHandlers.MaxAmount)
+	rs.AddLocalFunc("validate_amount", ussdHandlers.ValidateAmount)
+	rs.AddLocalFunc("reset_transaction_amount", ussdHandlers.ResetTransactionAmount)
+	rs.AddLocalFunc("get_recipient", ussdHandlers.GetRecipient)
+	rs.AddLocalFunc("get_sender", ussdHandlers.GetSender)
+	rs.AddLocalFunc("get_amount", ussdHandlers.GetAmount)
+	rs.AddLocalFunc("reset_incorrect", ussdHandlers.ResetIncorrectPin)
+	rs.AddLocalFunc("save_firstname", ussdHandlers.SaveFirstname)
+	rs.AddLocalFunc("save_familyname", ussdHandlers.SaveFamilyname)
+	rs.AddLocalFunc("save_gender", ussdHandlers.SaveGender)
+	rs.AddLocalFunc("save_location", ussdHandlers.SaveLocation)
+	rs.AddLocalFunc("save_yob", ussdHandlers.SaveYob)
+	rs.AddLocalFunc("save_offerings", ussdHandlers.SaveOfferings)
+	rs.AddLocalFunc("quit_with_balance", ussdHandlers.QuitWithBalance)
+	rs.AddLocalFunc("reset_account_authorized", ussdHandlers.ResetAccountAuthorized)
+	rs.AddLocalFunc("reset_allow_update", ussdHandlers.ResetAllowUpdate)
+	rs.AddLocalFunc("get_profile_info", ussdHandlers.GetProfileInfo)
+	rs.AddLocalFunc("verify_yob", ussdHandlers.VerifyYob)
+	rs.AddLocalFunc("reset_incorrect_date_format", ussdHandlers.ResetIncorrectYob)
+	rs.AddLocalFunc("set_reset_single_edit", ussdHandlers.SetResetSingleEdit)
+	rs.AddLocalFunc("initiate_transaction", ussdHandlers.InitiateTransaction)
+
+	return ussdHandlers, nil
+}
+
+func ensureDbDir(dbDir string) error {
+	err := os.MkdirAll(dbDir, 0700)
+	if err != nil {
+		return fmt.Errorf("state dir create exited with error: %v\n", err)
+	}
+	return nil
+}
+
+func getStateStore(dbDir string, ctx context.Context) (db.Db, error) {
+	store := gdbmdb.NewGdbmDb()
+	storeFile := path.Join(dbDir, "state.gdbm")
+	store.Connect(ctx, storeFile)
+	return store, nil
+}
+
+func getUserdataDb(dbDir string, ctx context.Context) db.Db {
+	store := gdbmdb.NewGdbmDb()
+	storeFile := path.Join(dbDir, "userdata.gdbm")
+	store.Connect(ctx, storeFile)
+	
+	return store
+}
+
+func getResource(resourceDir string, ctx context.Context) (resource.Resource, error) {
+	store := fsdb.NewFsDb()
+	err := store.Connect(ctx, resourceDir)
+	if err != nil {
+		return nil, err
+	}
+	rfs := resource.NewDbResource(store)
+	return rfs, nil
+}
+
+
+func main() {
+	var dbDir string
+	var resourceDir string
+	var size uint
+	var engineDebug bool
+	var stateDebug bool
+	var host string
+	var port uint
+	flag.StringVar(&dbDir, "dbdir", ".state", "database dir to read from")
+	flag.StringVar(&resourceDir, "resourcedir", path.Join("services", "registration"), "resource dir")
+	flag.BoolVar(&engineDebug, "engine-debug", false, "use engine debug output")
+	flag.BoolVar(&stateDebug, "state-debug", false, "use engine debug output")
+	flag.UintVar(&size, "s", 160, "max size of output")
+	flag.StringVar(&host, "h", "127.0.0.1", "http host")
+	flag.UintVar(&port, "p", 7123, "http port")
+	flag.Parse()
+
+	logg.Infof("start command", "dbdir", dbDir, "resourcedir", resourceDir,  "outputsize", size)
+
+	ctx := context.Background()
+	pfp := path.Join(scriptDir, "pp.csv")
+	flagParser, err := getFlags(pfp, true)
+
+	if err != nil {
+		os.Exit(1)
+	}
+
+	cfg := engine.Config{
+		Root:       "root",
+		OutputSize: uint32(size),
+		FlagCount:  uint32(16),
+	}
+	if stateDebug {
+		cfg.StateDebug = true
+	}
+	if engineDebug {
+		cfg.EngineDebug = true
+	}
+
+	rs, err := getResource(resourceDir, ctx)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, err.Error())
+		os.Exit(1)
+	}
+
+	err = ensureDbDir(dbDir)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, err.Error())
+		os.Exit(1)
+	}
+
+	userdataStore := getUserdataDb(dbDir, ctx)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, err.Error())
+		os.Exit(1)
+	}
+	defer userdataStore.Close()
+
+	dbResource, ok := rs.(*resource.DbResource)
+	if !ok {
+		os.Exit(1)
+	}
+
+	hl, err := getHandler(flagParser, dbResource, userdataStore)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, err.Error())
+		os.Exit(1)
+	}
+
+	stateStore, err := getStateStore(dbDir, ctx)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, err.Error())
+		os.Exit(1)
+	}
+	defer stateStore.Close()
+
+	rp := &atRequestParser{}
+	bsh := handlers.NewBaseSessionHandler(cfg, rs, stateStore, userdataStore, rp, hl)
+	sh := httpserver.NewATSessionHandler(bsh)
+	s := &http.Server{
+		Addr: fmt.Sprintf("%s:%s", host, strconv.Itoa(int(port))),
+		Handler: sh,
+	}
+	s.RegisterOnShutdown(sh.Shutdown)
+
+	cint := make(chan os.Signal)
+	cterm := make(chan os.Signal)
+	signal.Notify(cint, os.Interrupt, syscall.SIGINT)
+	signal.Notify(cterm, os.Interrupt, syscall.SIGTERM)
+	go func() {
+		select {
+		case _ = <-cint:
+		case _ = <-cterm:
+		}
+		s.Shutdown(ctx)
+	}()
+	err = s.ListenAndServe()
+	if err != nil {
+		logg.Infof("Server closed with error", "err", err)
+	}
+}

internal/handlers/base.go

@@ -1,13 +1,13 @@
 package handlers
 
 import (
-	"git.defalsify.org/vise.git/engine"
-	"git.defalsify.org/vise.git/resource"
-	"git.defalsify.org/vise.git/persist"
 	"git.defalsify.org/vise.git/db"
+	"git.defalsify.org/vise.git/engine"
+	"git.defalsify.org/vise.git/persist"
+	"git.defalsify.org/vise.git/resource"
 
-	"git.grassecon.net/urdt/ussd/internal/storage"
 	"git.grassecon.net/urdt/ussd/internal/handlers/ussd"
+	"git.grassecon.net/urdt/ussd/internal/storage"
 )
 
 type BaseSessionHandler struct {
@@ -93,7 +93,7 @@ func(f *BaseSessionHandler) Process(rqs RequestSession) (RequestSession, error)
 		return rqs, err
 	}
 
-	_ = r
+	rqs.Continue = r 
 	return rqs, nil
 }
 

internal/handlers/single.go

@@ -34,6 +34,7 @@ type RequestSession struct {
 	Input []byte
 	Storage *storage.Storage
 	Writer io.Writer
+	Continue bool
 }
 
 type engineMaker func(cfg engine.Config, rs resource.Resource, pr *persist.Persister) engine.Engine

internal/http/at_session_handler.go

@@ -0,0 +1,93 @@
+package http
+
+import (
+	"io"
+	"net/http"
+
+	"git.grassecon.net/urdt/ussd/internal/handlers"
+)
+
+type ATSessionHandler struct {
+	*SessionHandler
+}
+
+func NewATSessionHandler(h handlers.RequestHandler) *ATSessionHandler {
+	return &ATSessionHandler{
+		SessionHandler: ToSessionHandler(h),
+	}
+}
+
+func (ash *ATSessionHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
+	var code int
+	var err error
+
+	rqs := handlers.RequestSession{
+		Ctx:    req.Context(),
+		Writer: w,
+	}
+
+	rp := ash.GetRequestParser()
+	cfg := ash.GetConfig()
+	cfg.SessionId, err = rp.GetSessionId(req)
+	if err != nil {
+		logg.ErrorCtxf(rqs.Ctx, "", "header processing error", err)
+		ash.writeError(w, 400, err)
+	}
+	rqs.Config = cfg
+	rqs.Input, err = rp.GetInput(req)
+	if err != nil {
+		logg.ErrorCtxf(rqs.Ctx, "", "header processing error", err)
+		ash.writeError(w, 400, err)
+		return
+	}
+
+	rqs, err = ash.Process(rqs)
+	switch err {
+	case handlers.ErrStorage:
+		code = 500
+	case handlers.ErrEngineInit:
+		code = 500
+	case handlers.ErrEngineExec:
+		code = 500
+	default:
+		code = 200
+	}
+
+	if code != 200 {
+		ash.writeError(w, 500, err)
+		return
+	}
+
+	w.WriteHeader(200)
+	w.Header().Set("Content-Type", "text/plain")
+	rqs, err = ash.Output(rqs)
+	if err != nil {
+		ash.writeError(w, 500, err)
+		return
+	}
+
+	rqs, err = ash.Reset(rqs)
+	if err != nil {
+		ash.writeError(w, 500, err)
+		return
+	}
+}
+
+func (ash *ATSessionHandler) Output(rqs handlers.RequestSession) (handlers.RequestSession, error) {
+	var err error
+	var prefix string
+
+	if rqs.Continue {
+		prefix = "CON "
+	} else {
+		prefix = "END "
+	}
+
+	_, err = io.WriteString(rqs.Writer, prefix)
+	if err != nil {
+		return rqs, err
+	}
+
+	_, err = rqs.Engine.WriteResult(rqs.Ctx, rqs.Writer)
+	return rqs, err
+}