coin distribution preparation

.github/workflows/CICD.yaml

@@ -175,7 +175,7 @@ jobs:
     name: Test
     strategy:
       matrix:
-        package: [ "miner", "extra-bonus-notifier", "tokenomics", "cmd/freezer", "cmd/freezer-refrigerant", "cmd/freezer-miner"]
+        package: [ "miner", "coin-distribution", "extra-bonus-notifier", "tokenomics", "cmd/freezer", "cmd/freezer-refrigerant", "cmd/freezer-miner", "cmd/freezer-coin-distributer"]
     if: ${{ (github.event_name == 'pull_request' && github.event.pull_request.draft == false) || github.event_name == 'push'  }}
     runs-on: ubuntu-latest
     #    runs-on: self-hosted-ubuntu-latest-x64
@@ -226,7 +226,7 @@ jobs:
     name: Benchmark
     strategy:
       matrix:
-        package: [ "miner", "extra-bonus-notifier", "tokenomics", "cmd/freezer", "cmd/freezer-refrigerant", "cmd/freezer-miner"]
+        package: [ "miner", "coin-distribution", "extra-bonus-notifier", "tokenomics", "cmd/freezer", "cmd/freezer-refrigerant", "cmd/freezer-miner", "cmd/freezer-coin-distributer"]
     if: ${{ (github.event_name == 'pull_request' && github.event.pull_request.draft == false) || github.event_name == 'push'  }}
     runs-on: ubuntu-latest
     #    runs-on: self-hosted-ubuntu-latest-x64
@@ -276,7 +276,7 @@ jobs:
     name: Verify Dockerfile
     strategy:
       matrix:
-        service: [ "freezer", "freezer-refrigerant", "freezer-miner"]
+        service: [ "freezer", "freezer-refrigerant", "freezer-miner", "freezer-coin-distributer"]
         #those are not supported by golang docker image: linux/riscv64
         #platforms: linux/s390x,linux/arm64,linux/amd64,linux/ppc64le
         #commented because build takes too damn much with the other 3 platforms (~10 mins for each!!!) and we don`t need them atm
@@ -422,7 +422,7 @@ jobs:
     name: Push Docker
     strategy:
       matrix:
-        service: [ "freezer", "freezer-refrigerant", "freezer-miner"]
+        service: [ "freezer", "freezer-refrigerant", "freezer-miner", "freezer-coin-distributer"]
         #those are not supported by golang docker image: linux/riscv64
         #platforms: linux/s390x,linux/arm64,linux/amd64,linux/ppc64le
         #commented because build takes too damn much with the other 3 platforms (~10 mins for each!!!) and we don`t need them atm
@@ -528,6 +528,12 @@ jobs:
           cmd: |
             cd secret-infrastructure
             yq e -i '.generic-service-chart.applicationImage.tag = strenv(APP_TAG)' helm/freezer-miner/staging/common-values.yaml
+      - name: Update [staging] application tag version in helm/freezer-coin-distributer/staging/common-values.yaml
+        uses: mikefarah/yq@master
+        with:
+          cmd: |
+            cd secret-infrastructure
+            yq e -i '.generic-service-chart.applicationImage.tag = strenv(APP_TAG)' helm/freezer-coin-distributer/staging/common-values.yaml
       - name: Commit and Push Changes to Application Tag Version
         run: |
           cd secret-infrastructure
@@ -536,7 +542,8 @@ jobs:
           git add helm/freezer/staging/common-values.yaml 
           git add helm/freezer-refrigerant/staging/common-values.yaml 
           git add helm/freezer-miner/staging/common-values.yaml 
-          git commit -m "Updated 'freezer' & 'freezer-refrigerant' & 'freezer-miner' tag version (${{env.APP_TAG}}) in application helm chart deployment manifests"
+          git add helm/freezer-coin-distributer/staging/common-values.yaml 
+          git commit -m "Updated 'freezer' & 'freezer-refrigerant' & 'freezer-miner' & 'freezer-coin-distributer' tag version (${{env.APP_TAG}}) in application helm chart deployment manifests"
           git push --set-upstream origin master
       - name: Slack Notification For Success
         if: ${{ success() }}

application.yaml

@@ -180,6 +180,9 @@ miner:
   batchSize: 100
   wintr/connectors/storage/v2: *db
 coin-distribution:
+  alert-slack-webhook: https://hooks.slack.com/services/dummy/dummy/dummy
+  environment: local
+  review-url: https://some.bogus.example.com/going/somewhere
   development: true
   workers: 2
   batchSize: 100

cmd/freezer/api/docs.go

@@ -776,6 +776,10 @@ const docTemplate = `{
                     "type": "string",
                     "example": "1,243.02"
                 },
+                "totalMiningBlockchain": {
+                    "type": "string",
+                    "example": "1,243.02"
+                },
                 "totalNoPreStakingBonus": {
                     "type": "string",
                     "example": "1,243.02"

cmd/freezer/api/swagger.json

@@ -770,6 +770,10 @@
                     "type": "string",
                     "example": "1,243.02"
                 },
+                "totalMiningBlockchain": {
+                    "type": "string",
+                    "example": "1,243.02"
+                },
                 "totalNoPreStakingBonus": {
                     "type": "string",
                     "example": "1,243.02"

cmd/freezer/api/swagger.yaml

@@ -78,6 +78,9 @@ definitions:
       total:
         example: 1,243.02
         type: string
+      totalMiningBlockchain:
+        example: 1,243.02
+        type: string
       totalNoPreStakingBonus:
         example: 1,243.02
         type: string

coin-distribution/.testdata/application.yaml

@@ -5,6 +5,9 @@ logger:
   encoder: console
   level: debug
 coin-distribution:
+  alert-slack-webhook: https://hooks.slack.com/services/dummy/dummy/dummy
+  environment: local
+  review-url: https://some.bogus.example.com/going/somewhere
   startHours: 12
   endHours: 17
   development: true

coin-distribution/DDL.sql

@@ -7,22 +7,37 @@ EXCEPTION
     WHEN duplicate_object THEN null;
 END $$;
 
+DO $$
+BEGIN
+    IF NOT EXISTS (SELECT 1 FROM pg_type WHERE typname = 'pending_coin_distributions_status') THEN
+        create type pending_coin_distributions_status AS ENUM ('NEW', 'PENDING', 'ACCEPTED', 'REJECTED');
+    END IF;
+END
+$$;
+
+
 CREATE TABLE IF NOT EXISTS pending_coin_distributions  (
                     created_at                timestamp NOT NULL,
                     internal_id               bigint    NOT NULL,
                     day                       date      NOT NULL,
                     iceflakes                 uint256,
                     user_id                   text      NOT NULL,
                     eth_address               text      NOT NULL,
-                    PRIMARY KEY(day, user_id));
+                    eth_status                pending_coin_distributions_status NOT NULL DEFAULT 'NEW',
+                    eth_tx                    text,
+                    PRIMARY KEY(day, user_id))
+                    WITH (FILLFACTOR = 70);
 
-CREATE INDEX IF NOT EXISTS pending_coin_distributions_worker_number_ix ON pending_coin_distributions ((internal_id % 10), created_at ASC);
+CREATE INDEX IF NOT EXISTS pending_coin_distributions_worker_number_ix ON pending_coin_distributions (eth_status, (internal_id % 10), created_at ASC);
+CREATE INDEX IF NOT EXISTS pending_coin_distributions_eth_status_tx_ix ON pending_coin_distributions (eth_status, eth_tx);
 
 CREATE TABLE IF NOT EXISTS global (
                     key       text NOT NULL primary key,
-                    value     text NOT NULL );
+                    value     text NOT NULL )
+                    WITH (FILLFACTOR = 70);
 INSERT INTO global (key,value)
-            VALUES ('coin_distributer_enabled','true')
+            VALUES ('coin_distributer_enabled','true'),
+                   ('coin_collector_enabled','true')
          ON CONFLICT(key) DO NOTHING;
 
 CREATE TABLE IF NOT EXISTS coin_distributions_by_earner (
@@ -35,7 +50,8 @@ CREATE TABLE IF NOT EXISTS coin_distributions_by_earner (
                     user_id                   text      NOT NULL,
                     earner_user_id            text      NOT NULL,
                     eth_address               text      NOT NULL,
-                    PRIMARY KEY(day, user_id, earner_user_id));
+                    PRIMARY KEY(day, user_id, earner_user_id))
+                    WITH (FILLFACTOR = 70);
 
 CREATE TABLE IF NOT EXISTS coin_distributions_pending_review  (
                     created_at                timestamp NOT NULL,
@@ -75,4 +91,77 @@ CREATE TABLE IF NOT EXISTS reviewed_coin_distributions  (
                     eth_address               text      NOT NULL,
                     reviewer_user_id          text      NOT NULL,
                     decision                  text      NOT NULL,
-                    PRIMARY KEY(user_id, day, review_day));
+                    PRIMARY KEY(user_id, day, review_day));
+
+create or replace procedure approve_coin_distributions(reviewer_user_id text, nested boolean)
+language plpgsql
+    as $$
+declare
+         now timestamp := current_timestamp;
+BEGIN
+    insert into pending_coin_distributions(created_at, internal_id, day, iceflakes, user_id, eth_address)
+    select created_at, internal_id, day, iceflakes, user_id, eth_address
+    from coin_distributions_pending_review;
+
+    insert into reviewed_coin_distributions(reviewed_at, created_at, internal_id, ice, day, review_day, iceflakes, username, referred_by_username, user_id, eth_address, reviewer_user_id, decision)
+    select now, created_at, internal_id, ice, day, now::date, iceflakes, username, referred_by_username, user_id, eth_address, reviewer_user_id, 'approve'
+    from coin_distributions_pending_review;
+
+    delete from coin_distributions_pending_review where 1=1;
+
+    IF nested is false THEN
+        commit;
+    END IF;
+end; $$;
+
+create or replace procedure deny_coin_distributions(reviewer_user_id text, nested boolean)
+language plpgsql
+    as $$
+declare
+         now timestamp := current_timestamp;
+BEGIN
+    insert into reviewed_coin_distributions(reviewed_at, created_at, internal_id, ice, day, review_day, iceflakes, username, referred_by_username, user_id, eth_address, reviewer_user_id, decision)
+    select now, created_at, internal_id, ice, day, now::date, iceflakes, username, referred_by_username, user_id, eth_address, reviewer_user_id, 'deny'
+    from coin_distributions_pending_review;
+
+    delete from coin_distributions_pending_review where 1=1;
+
+    INSERT INTO global (key,value)
+                VALUES ('coin_distributer_enabled','false'),
+                       ('coin_collector_enabled','false')
+    ON CONFLICT (key) DO UPDATE
+    			SET value = EXCLUDED.value;
+
+    IF nested is false THEN
+        commit;
+    END IF;
+end; $$;
+
+create or replace procedure prepare_coin_distributions_for_review(nested boolean)
+language plpgsql
+    as $$
+declare
+         zeros text := '0000000000000000';
+BEGIN
+    delete from coin_distributions_by_earner WHERE balance = 0;
+
+    insert into coin_distributions_pending_review(created_at, internal_id, ice, day, iceflakes, username, referred_by_username, user_id, eth_address)
+        SELECT created_at, internal_id, ice, day, (ice::text||zeros)::uint256 AS iceflakes, username, referred_by_username, user_id, eth_address
+        FROM (select
+                   min (created_at) filter ( where user_id=earner_user_id )  AS created_at,
+                   min (internal_id) filter ( where user_id=earner_user_id )  AS internal_id,
+                   sum(balance) AS ice,
+                   min (day) filter ( where user_id=earner_user_id ) AS day,
+                   string_agg(username,'') AS username,
+                   string_agg(referred_by_username,'') AS referred_by_username,
+                   user_id,
+                   string_agg(eth_address,'') AS eth_address
+                from coin_distributions_by_earner
+                group by day,user_id) AS X;
+
+    delete from coin_distributions_by_earner where 1=1;
+
+    IF nested is false THEN
+        commit;
+    END IF;
+end; $$;

coin-distribution/coin_distribution.go

@@ -26,25 +26,12 @@ func MustStartCoinDistribution(ctx context.Context, cancel context.CancelFunc) C
 		db: storage.MustConnect(context.Background(), ddl, applicationYamlKey),
 		wg: new(sync.WaitGroup),
 	}
-	cd.wg.Add(int(cfg.Workers))
 	cd.cancel = cancel
-
-	for workerNumber := int64(0); workerNumber < cfg.Workers; workerNumber++ {
-		go func(wn int64) {
-			defer cd.wg.Done()
-			cd.distributeCoins(ctx, wn)
-		}(workerNumber)
-	}
+	go startPrepareCoinDistributionsForReviewMonitor(ctx, cd.db)
 
 	return cd
 }
 
-func NewRepository(ctx context.Context, _ context.CancelFunc) Repository {
-	repo := &repository{db: storage.MustConnect(ctx, ddl, applicationYamlKey)}
-
-	return repo
-}
-
 func (cd *coinDistributer) Close() error {
 	cd.cancel()
 	cd.wg.Wait()
@@ -123,10 +110,3 @@ func (cd *coinDistributer) Disable(rooCtx context.Context) error {
 
 	return nil
 }
-
-func (r *repository) CheckHealth(ctx context.Context) error {
-	return errors.Wrap(r.db.Ping(ctx), "[health-check] failed to ping DB for coindistribution.repository")
-}
-func (r *repository) Close() error {
-	return errors.Wrap(r.db.Close(), "failed to close db")
-}

coin-distribution/contract.go

@@ -26,6 +26,9 @@ type (
 		GetCoinDistributionsForReview(ctx context.Context, arg *GetCoinDistributionsForReviewArg) (*CoinDistributionsForReview, error)
 		CheckHealth(ctx context.Context) error
 		ReviewCoinDistributions(ctx context.Context, reviewerUserID string, decision string) error
+		NotifyCoinDistributionCollectionCycleEnded(ctx context.Context) error
+		GetCollectorStatus(ctx context.Context) (latestCollectingDate *time.Time, collectorEnabled bool, err error)
+		CollectCoinDistributionsForReview(ctx context.Context, records []*ByEarnerForReview) error
 	}
 
 	CoinDistributionsForReview struct {
@@ -56,6 +59,17 @@ type (
 		Ice                float64    `json:"ice" db:"-" example:"1000"`
 		IceInternal        int64      `json:"-" db:"ice" swaggerignore:"true"`
 	}
+
+	ByEarnerForReview struct {
+		CreatedAt          *time.Time
+		Username           string
+		ReferredByUsername string
+		UserID             string
+		EarnerUserID       string
+		EthAddress         string
+		InternalID         int64
+		Balance            float64
+	}
 )
 
 // Private API.
@@ -80,13 +94,17 @@ type (
 		wg     *sync.WaitGroup
 	}
 	repository struct {
-		db *storage.DB
+		cfg *config
+		db  *storage.DB
 	}
 	config struct {
-		StartHours  int   `yaml:"startHours"`
-		EndHours    int   `yaml:"endHours"`
-		Workers     int64 `yaml:"workers"`
-		BatchSize   int64 `yaml:"batchSize"`
-		Development bool  `yaml:"development"`
+		AlertSlackWebhook string `yaml:"alert-slack-webhook" mapstructure:"alert-slack-webhook"` //nolint:tagliatelle // .
+		Environment       string `yaml:"environment" mapstructure:"environment"`
+		ReviewURL         string `yaml:"review-url" mapstructure:"review-url"`
+		StartHours        int    `yaml:"startHours"`
+		EndHours          int    `yaml:"endHours"`
+		Workers           int64  `yaml:"workers"`
+		BatchSize         int64  `yaml:"batchSize"`
+		Development       bool   `yaml:"development"`
 	}
 )