Add outdated operations kind (#5)

Author: aopoltorzhicky

build/dipdup.mainnet.yml

@@ -40,6 +40,8 @@ hasura:
   admin_secret: ${ADMIN_SECRET:-changeme}
   select_limit: 100
   allow_aggregation: false
+  source: default
+  add_source: true
   rest: true
 
 prometheus:

build/dipdup.testnet.yml

@@ -62,6 +62,8 @@ hasura:
   admin_secret: ${ADMIN_SECRET:-changeme}
   select_limit: 100
   allow_aggregation: false
+  source: default
+  add_source: true
   rest: true
 
 prometheus:

cmd/mempool/indexer.go

@@ -270,7 +270,7 @@ func (indexer *Indexer) listen(ctx context.Context) {
 					log.Err(err).Msg("handleAppliedOperation")
 					continue
 				}
-			case receiver.StatusBranchDelayed, receiver.StatusBranchRefused, receiver.StatusRefused, receiver.StatusUnprocessed:
+			case receiver.StatusBranchDelayed, receiver.StatusBranchRefused, receiver.StatusRefused, receiver.StatusUnprocessed, receiver.StatusOutdated:
 				failed, ok := msg.Body.(node.FailedMonitor)
 				if !ok {
 					indexer.error().Msgf("invalid %s operation %v", msg.Status, failed)

cmd/mempool/main.go

@@ -121,8 +121,12 @@ func main() {
 		for kind := range kinds {
 			t = append(t, kind)
 		}
-		tables := models.GetModelsBy(t...)
-		if err := hasura.Create(ctx, cfg.Hasura, cfg.Database, views, tables...); err != nil {
+		if err := hasura.Create(ctx, hasura.GenerateArgs{
+			Config:         cfg.Hasura,
+			DatabaseConfig: cfg.Database,
+			Views:          views,
+			Models:         models.GetModelsBy(t...),
+		}); err != nil {
 			log.Err(err).Msg("")
 			cancel()
 			return

cmd/mempool/receiver/data.go

@@ -16,5 +16,6 @@ const (
 	StatusBranchDelayed Status = "branch_delayed"
 	StatusBranchRefused Status = "branch_refused"
 	StatusRefused       Status = "refused"
+	StatusOutdated      Status = "outdated"
 	StatusUnprocessed   Status = "unprocessed"
 )

cmd/mempool/receiver/receiver.go

@@ -77,6 +77,7 @@ func (indexer *Receiver) Start(ctx context.Context) {
 		indexer.monitors[i].SubscribeOnMempoolBranchDelayed(ctx)
 		indexer.monitors[i].SubscribeOnMempoolBranchRefused(ctx)
 		indexer.monitors[i].SubscribeOnMempoolRefused(ctx)
+		indexer.monitors[i].SubscribeOnMempoolOutdated(ctx)
 	}
 }
 
@@ -137,6 +138,14 @@ func (indexer *Receiver) run(ctx context.Context, monitor *node.Monitor) {
 					Protocol: indexer.protocol,
 				}
 			}
+		case outdated := <-monitor.Outdated():
+			for i := range outdated {
+				indexer.operations <- Message{
+					Status:   StatusOutdated,
+					Body:     *outdated[i],
+					Protocol: indexer.protocol,
+				}
+			}
 		}
 	}
 }

go.mod

@@ -4,7 +4,7 @@ go 1.18
 
 require (
 	github.com/btcsuite/btcutil v1.0.2
-	github.com/dipdup-net/go-lib v0.2.5
+	github.com/dipdup-net/go-lib v0.2.11
 	github.com/go-pg/pg/v10 v10.10.6
 	github.com/karlseguin/ccache v2.0.3+incompatible
 	github.com/mitchellh/mapstructure v1.4.2

go.sum

@@ -38,8 +38,8 @@ github.com/davecgh/go-spew v0.0.0-20171005155431-ecdeabc65495/go.mod h1:J7Y8YcW2
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/dipdup-net/go-lib v0.2.5 h1:9CAcLmbD+0gcJ1S3g5ZPHulnV8x/GAIP1f0RMZ/VZNQ=
-github.com/dipdup-net/go-lib v0.2.5/go.mod h1:RjVhNv63RE3pOAeHLwnspa2UjJi80k/Kelwv7aN1ETE=
+github.com/dipdup-net/go-lib v0.2.11 h1:ZzPMOOa4y03UyC/7BbXKII5Ue/i76QvVavG4TArMSlg=
+github.com/dipdup-net/go-lib v0.2.11/go.mod h1:RjVhNv63RE3pOAeHLwnspa2UjJi80k/Kelwv7aN1ETE=
 github.com/ebellocchia/go-base58 v0.1.0 h1:0w/ODEfZnOPW5KW0QY/Xpb1fxba/BxQJMUa5iYzpljk=
 github.com/ebellocchia/go-base58 v0.1.0/go.mod h1:RHE/6C6Ru6YAH9Tc+A9eHQ6ZKEooLC0jw+YLnpt3CAU=
 github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=