Merge pull request #45 from rarimo/feat/points-for-early-test

Feat/points for early test

internal/cli/event.go

@@ -0,0 +1,89 @@
+package cli
+
+import (
+	"fmt"
+
+	"github.com/rarimo/rarime-points-svc/internal/config"
+	"github.com/rarimo/rarime-points-svc/internal/data"
+	"github.com/rarimo/rarime-points-svc/internal/data/evtypes"
+	"github.com/rarimo/rarime-points-svc/internal/data/pg"
+	"github.com/rarimo/rarime-points-svc/internal/service/handlers"
+)
+
+func emitEvent(cfg config.Config, timestamp int) {
+	log := cfg.Log()
+	db := cfg.DB()
+	lvls := cfg.Levels()
+	evTypes := cfg.EventTypes()
+
+	balancesQ := pg.NewBalances(db)
+	eventsQ := pg.NewEvents(db)
+	referralsQ := pg.NewReferrals(db)
+	countriesQ := pg.NewCountries(db)
+
+	evType := evTypes.Get(evtypes.TypeEarlyTest, evtypes.FilterInactive)
+
+	if evType == nil {
+		log.Infof("Event type %s is inactive", evtypes.TypeEarlyTest)
+		return
+	}
+
+	balances, err := balancesQ.FilterByCreatedAtBefore(timestamp).FilterUnverified().Select()
+	if err != nil {
+		panic(fmt.Errorf("failed to select balances for early test reward: %w", err))
+	}
+	if len(balances) == 0 {
+		log.Infof("No balances found")
+		return
+	}
+
+	nullifiers := make([]string, 0, len(balances))
+	for _, balance := range balances {
+		nullifiers = append(nullifiers, balance.Nullifier)
+	}
+
+	emittedEvents, err := eventsQ.New().FilterByType(evtypes.TypeEarlyTest).FilterByNullifier(nullifiers...).Select()
+	if err != nil {
+		panic(fmt.Errorf("failed to select emitted events: %w", err))
+	}
+
+	eventsMap := make(map[string]struct{}, len(emittedEvents))
+	for _, event := range emittedEvents {
+		eventsMap[event.Nullifier] = struct{}{}
+	}
+
+	for _, balance := range balances {
+		err = eventsQ.New().Transaction(func() error {
+			if _, exists := eventsMap[balance.Nullifier]; exists {
+				log.Infof("Event %s is already done for user with nullifier %s ", evtypes.TypeEarlyTest, balance.Nullifier)
+				return nil
+			}
+
+			err = eventsQ.Insert(data.Event{
+				Nullifier: balance.Nullifier,
+				Type:      evtypes.TypeEarlyTest,
+				Status:    data.EventFulfilled,
+			})
+
+			if err != nil {
+				return fmt.Errorf("failed to insert %s event: %w", evtypes.TypeEarlyTest, err)
+			}
+
+			if !evType.AutoClaim {
+				return nil
+			}
+
+			_, err = eventsQ.FilterByNullifier(balance.Nullifier).Update(data.EventClaimed, nil, &evType.Reward)
+			if err != nil {
+				return fmt.Errorf("failed to update %s events for user=%s: %w", evtypes.TypeEarlyTest, balance.Nullifier, err)
+			}
+
+			err := handlers.DoClaimEventUpdates(lvls, referralsQ, balancesQ, countriesQ, balance, evType.Reward)
+			if err != nil {
+				return fmt.Errorf("failed to do lvlup and referrals updates: %w", err)
+			}
+
+			return nil
+		})
+	}
+}

internal/cli/main.go

@@ -29,6 +29,11 @@ func Run(args []string) bool {
 		migrateCmd     = app.Command("migrate", "migrate command")
 		migrateUpCmd   = migrateCmd.Command("up", "migrate db up")
 		migrateDownCmd = migrateCmd.Command("down", "migrate db down")
+
+		eventCmd     = app.Command("events", "manage events")
+		eventEmitCmd = eventCmd.Command("emit", "emit event")
+
+		onboarderBefore = eventEmitCmd.Arg("before", "balance onboarded before this timestamp").Required().Int()
 	)
 
 	cmd, err := app.Parse(args[1:])
@@ -48,6 +53,8 @@ func Run(args []string) bool {
 		err = MigrateUp(cfg)
 	case migrateDownCmd.FullCommand():
 		err = MigrateDown(cfg)
+	case eventEmitCmd.FullCommand():
+		emitEvent(cfg, *onboarderBefore)
 	default:
 		log.Errorf("unknown command %s", cmd)
 		return false

internal/data/balances.go

@@ -47,9 +47,11 @@ type BalancesQ interface {
 	WithoutPassportEvent() ([]WithoutPassportEventBalance, error)
 	WithoutReferralEvent() ([]ReferredReferrer, error)
 
+	FilterByCreatedAtBefore(timestamp int) BalancesQ
 	FilterByNullifier(...string) BalancesQ
 	FilterDisabled() BalancesQ
 	FilterByAnonymousID(id string) BalancesQ
+	FilterUnverified() BalancesQ
 }
 
 type WithoutPassportEventBalance struct {

internal/data/events.go

@@ -58,7 +58,7 @@ type EventsQ interface {
 	SelectAbsentTypes(allTypes ...string) ([]ReopenableEvent, error)
 
 	FilterByID(...string) EventsQ
-	FilterByNullifier(string) EventsQ
+	FilterByNullifier(...string) EventsQ
 	FilterByStatus(...EventStatus) EventsQ
 	FilterByType(...string) EventsQ
 	FilterByNotType(types ...string) EventsQ

internal/data/evtypes/main.go

@@ -27,6 +27,7 @@ const (
 	TypeBeReferred       = "be_referred"
 	TypeReferralSpecific = "referral_specific"
 	TypePassportScan     = "passport_scan"
+	TypeEarlyTest        = "early_test"
 )
 
 const (

internal/data/pg/balances.go

@@ -215,6 +215,14 @@ func (q *balances) FilterByAnonymousID(id string) data.BalancesQ {
 	return q.applyCondition(squirrel.Eq{"anonymous_id": id})
 }
 
+func (q *balances) FilterByCreatedAtBefore(timestamp int) data.BalancesQ {
+	return q.applyCondition(squirrel.LtOrEq{"created_at": timestamp})
+}
+
+func (q *balances) FilterUnverified() data.BalancesQ {
+	return q.applyCondition(squirrel.NotEq{"anonymous_id": nil})
+}
+
 func (q *balances) applyCondition(cond squirrel.Sqlizer) data.BalancesQ {
 	q.selector = q.selector.Where(cond)
 	q.updater = q.updater.Where(cond)

internal/data/pg/events.go

@@ -190,7 +190,7 @@ func (q *events) FilterByID(ids ...string) data.EventsQ {
 	return q.applyCondition(squirrel.Eq{"id": ids})
 }
 
-func (q *events) FilterByNullifier(nullifier string) data.EventsQ {
+func (q *events) FilterByNullifier(nullifier ...string) data.EventsQ {
 	return q.applyCondition(squirrel.Eq{"nullifier": nullifier})
 }