configure pipeline filter for all wallet addresses

go.mod

@@ -8,18 +8,21 @@ require (
 	github.com/0xSplits/workit v0.7.1
 	github.com/google/gofuzz v1.2.0
 	github.com/gorilla/mux v1.8.1
+	github.com/jackc/pgx/v5 v5.7.6
 	github.com/joho/godotenv v1.5.1
 	github.com/kelseyhightower/envconfig v1.4.0
 	github.com/prometheus/client_golang v1.23.2
 	github.com/rs/cors v1.11.1
 	github.com/spf13/cobra v1.10.1
 	github.com/twitchtv/twirp v8.1.3+incompatible
+	github.com/xh3b4sd/choreo v0.6.0
 	github.com/xh3b4sd/logger v0.11.1
 	github.com/xh3b4sd/tracer v1.0.0
 	go.opentelemetry.io/otel/metric v1.38.0
 )
 
 require (
+	cirello.io/pglock v1.16.1
 	github.com/0xSplits/indexingo v0.1.0
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
@@ -28,7 +31,11 @@ require (
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
+	github.com/jackc/pgpassfile v1.0.0 // indirect
+	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
+	github.com/jackc/puddle/v2 v2.2.2 // indirect
 	github.com/klauspost/compress v1.18.1 // indirect
+	github.com/lib/pq v1.10.9 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/prometheus/client_model v0.6.2 // indirect
 	github.com/prometheus/common v0.67.2 // indirect
@@ -42,6 +49,9 @@ require (
 	go.opentelemetry.io/otel/sdk/metric v1.38.0 // indirect
 	go.opentelemetry.io/otel/trace v1.38.0 // indirect
 	go.yaml.in/yaml/v2 v2.4.3 // indirect
+	golang.org/x/crypto v0.38.0 // indirect
+	golang.org/x/sync v0.17.0 // indirect
 	golang.org/x/sys v0.37.0 // indirect
+	golang.org/x/text v0.30.0 // indirect
 	google.golang.org/protobuf v1.36.10 // indirect
 )

go.sum

@@ -1,3 +1,5 @@
+cirello.io/pglock v1.16.1 h1:NKVe133vhtJMnzdL0UlCtyEb9rm3vSnT2/R7MNu6mfA=
+cirello.io/pglock v1.16.1/go.mod h1:aCm2qCmp8jc2cNSFUW+n94Ie6ZGXmauBBSXVkWSKD/M=
 github.com/0xSplits/indexingo v0.1.0 h1:t4Bl+LpUTG3tpNAJG8JIMTAr4UOgo+JhanBwBBaRJhs=
 github.com/0xSplits/indexingo v0.1.0/go.mod h1:nS0gqM7Y3n++v/5OU2dTtjo5u1pAOKLNnk3KuX2Ic9g=
 github.com/0xSplits/otelgo v0.1.2 h1:QjbUMNNQcUsnkOmZ35bc3Fbhz7u0PA611LYrh4aOpPk=
@@ -6,13 +8,16 @@ github.com/0xSplits/pulsargocode v0.1.0 h1:wWHCxp+IfxZIJ6mbxnG2hG+R8Hrpu15lPbI6n
 github.com/0xSplits/pulsargocode v0.1.0/go.mod h1:lt3sg7Pvgam0AEOINEdoyHO2z80n4W7tm6XuDmm1kaA=
 github.com/0xSplits/workit v0.7.1 h1:ER2ITiqmahwoN94oz9XCUvdUJLnYB5yZyA9y8FAOwug=
 github.com/0xSplits/workit v0.7.1/go.mod h1:Ylbvjp0puOXWNVzYtI0cxFO37hqPNy/prwHIWau2gpQ=
+github.com/DATA-DOG/go-sqlmock v1.5.2 h1:OcvFkGmslmlZibjAjaHm3L//6LiuBgolP7OputlJIzU=
+github.com/DATA-DOG/go-sqlmock v1.5.2/go.mod h1:88MAG/4G7SMwSE3CeA0ZKzrT5CiOU3OJ+JlNzwDqpNU=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/coder/websocket v1.8.14 h1:9L0p0iKiNOibykf283eHkKUHHrpG7f65OE3BhhO7v9g=
 github.com/coder/websocket v1.8.14/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg=
 github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g=
+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/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
@@ -30,6 +35,14 @@ github.com/gorilla/mux v1.8.1 h1:TuBL49tXwgrFYWhqrNgrUNEY92u81SPhu7sTdzQEiWY=
 github.com/gorilla/mux v1.8.1/go.mod h1:AKf9I4AEqPTmMytcMc0KkNouC66V3BtZ4qD5fmWSiMQ=
 github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=
 github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
+github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
+github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
+github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo=
+github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
+github.com/jackc/pgx/v5 v5.7.6 h1:rWQc5FwZSPX58r1OQmkuaNicxdmExaEz5A2DO2hUuTk=
+github.com/jackc/pgx/v5 v5.7.6/go.mod h1:aruU7o91Tc2q2cFp5h4uP3f6ztExVpyVv88Xl/8Vl8M=
+github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo=
+github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
 github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0=
 github.com/joho/godotenv v1.5.1/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwAbqwq4=
 github.com/kelseyhightower/envconfig v1.4.0 h1:Im6hONhd3pLkfDFsbRgu68RDNkGF1r3dvMUtDTo2cv8=
@@ -42,6 +55,8 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc=
 github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
+github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw=
+github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
 github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
 github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
@@ -68,10 +83,15 @@ github.com/spf13/cobra v1.10.1/go.mod h1:7SmJGaTHFVBY0jW4NXGluQoLvhqFQM+6XSKD+P4
 github.com/spf13/pflag v1.0.9/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
 github.com/spf13/pflag v1.0.10 h1:4EBh2KAYBwaONj6b2Ye1GiHfwjqyROoF4RwYO+vPwFk=
 github.com/spf13/pflag v1.0.10/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
 github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
 github.com/twitchtv/twirp v8.1.3+incompatible h1:+F4TdErPgSUbMZMwp13Q/KgDVuI7HJXP61mNV3/7iuU=
 github.com/twitchtv/twirp v8.1.3+incompatible/go.mod h1:RRJoFSAmTEh2weEqWtpPE3vFK5YBhA6bqp2l1kfCC5A=
+github.com/xh3b4sd/choreo v0.6.0 h1:x6B/zPRwhwqaSt6eQ9bMom+9ROVGf728946hlcLwMjA=
+github.com/xh3b4sd/choreo v0.6.0/go.mod h1:FSmCDJYDGChWwn06JVitOrbFoSie8G5G34Zt4k16NBQ=
 github.com/xh3b4sd/logger v0.11.1 h1:aTK4ygh7aPv1jq54J8bx+zjH6A8RYdkKAgOZYw867C0=
 github.com/xh3b4sd/logger v0.11.1/go.mod h1:MC7Dp7RC3tZ182KlvSulGcRQVX/D2l+WlCSGLF1mvO8=
 github.com/xh3b4sd/tracer v1.0.0 h1:mr9uYCx/Ry2w1wdJz0V0Kq71/KeF+hUQjbZQJCxm3Zw=
@@ -94,12 +114,19 @@ go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
 go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
 go.yaml.in/yaml/v2 v2.4.3 h1:6gvOSjQoTB3vt1l+CU+tSyi/HOjfOjRLJ4YwYZGwRO0=
 go.yaml.in/yaml/v2 v2.4.3/go.mod h1:zSxWcmIDjOzPXpjlTTbAsKokqkDNAVtZO0WOMiT90s8=
+golang.org/x/crypto v0.38.0 h1:jt+WWG8IZlBnVbomuhg2Mdq0+BBQaHbtqHEFEigjUV8=
+golang.org/x/crypto v0.38.0/go.mod h1:MvrbAqul58NNYPKnOra203SB9vpuZW0e+RRZV+Ggqjw=
+golang.org/x/sync v0.17.0 h1:l60nONMj9l5drqw6jlhIELNv9I0A4OFgRsG9k2oT9Ug=
+golang.org/x/sync v0.17.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI=
 golang.org/x/sys v0.37.0 h1:fdNQudmxPjkdUTPnLn5mdQv7Zwvbvpaxqs831goi9kQ=
 golang.org/x/sys v0.37.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
+golang.org/x/text v0.30.0 h1:yznKA/E9zq54KzlzBEAWn1NXSQ8DIp/NYMy88xJjl4k=
+golang.org/x/text v0.30.0/go.mod h1:yDdHFIX9t+tORqspjENWgzaCVXgk0yYnYuSZ8UzzBVM=
 google.golang.org/protobuf v1.36.10 h1:AYd7cD/uASjIL6Q9LiTjz8JLcrh/88q5UObnmY3aOOE=
 google.golang.org/protobuf v1.36.10/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

pkg/daemon/daemon.go

@@ -1,6 +1,10 @@
 package daemon
 
 import (
+	"github.com/0xSplits/indexingo/client"
+	"github.com/0xSplits/indexingo/filters"
+	"github.com/0xSplits/indexingo/pipelines"
+	"github.com/0xSplits/indexingo/transformations"
 	"github.com/0xSplits/otelgo/recorder"
 	"github.com/0xSplits/pulsar/pkg/envvar"
 	"github.com/0xSplits/pulsar/pkg/runtime"
@@ -14,8 +18,11 @@ type Config struct {
 
 type Daemon struct {
 	env envvar.Env
+	fil *filters.Filters
 	log logger.Interface
 	met metric.Meter
+	pip *pipelines.Pipelines
+	tra *transformations.Transformations
 }
 
 func New(c Config) *Daemon {
@@ -35,6 +42,34 @@ func New(c Config) *Daemon {
 		})
 	}
 
+	var cli client.Interface
+	{
+		cli = client.New(client.Config{
+			Key: c.Env.IndexingcoApiKey,
+		})
+	}
+
+	var fil *filters.Filters
+	{
+		fil = filters.New(filters.Config{
+			Cli: cli,
+		})
+	}
+
+	var tra *transformations.Transformations
+	{
+		tra = transformations.New(transformations.Config{
+			Cli: cli,
+		})
+	}
+
+	var pip *pipelines.Pipelines
+	{
+		pip = pipelines.New(pipelines.Config{
+			Cli: cli,
+		})
+	}
+
 	log.Log(
 		"level", "info",
 		"message", "daemon is launching procs",
@@ -43,7 +78,10 @@ func New(c Config) *Daemon {
 
 	return &Daemon{
 		env: c.Env,
+		fil: fil,
 		log: log,
 		met: met,
+		pip: pip,
+		tra: tra,
 	}
 }

pkg/daemon/ensure.go

@@ -3,61 +3,11 @@ package daemon
 import (
 	"fmt"
 
-	"github.com/0xSplits/indexingo/client"
-	"github.com/0xSplits/indexingo/filters"
 	"github.com/0xSplits/indexingo/pipelines"
-	"github.com/0xSplits/indexingo/transformations"
 	"github.com/xh3b4sd/tracer"
 )
 
 func (d *Daemon) Ensure() error {
-	var cli client.Interface
-	{
-		cli = client.New(client.Config{
-			Key: d.env.IndexingcoApiKey,
-		})
-	}
-
-	//--------------------------------------------------------------------------//
-
-	var fil *filters.Filters
-	{
-		fil = filters.New(filters.Config{
-			Cli: cli,
-		})
-	}
-
-	var tra *transformations.Transformations
-	{
-		tra = transformations.New(transformations.Config{
-			Cli: cli,
-		})
-	}
-
-	var pip *pipelines.Pipelines
-	{
-		pip = pipelines.New(pipelines.Config{
-			Cli: cli,
-		})
-	}
-
-	//--------------------------------------------------------------------------//
-
-	{
-		res, err := fil.AddValues("test-filter", []string{"0xb7f5bf799fb265657c628ef4a13f90f83a3a616a"})
-		if err != nil {
-			tracer.Panic(tracer.Mask(err))
-		}
-
-		d.log.Log(
-			"level", "info",
-			"message", "filter creation",
-			"status", res.Message,
-		)
-	}
-
-	//--------------------------------------------------------------------------//
-
 	//
 	//     curl -s --location --globoff 'https://app.indexing.co/dw/transformations/test?network=base&beat=37740907&filter=xh3b4sd-test-filter&filterKeys[0]=to&filterKeys[1]=from' --header "x-api-key: $INDEXINGCO_API_KEY" --form 'code="function traByBlock(blo) { const tra = templates.tokenTransfers(blo); return tra.map(x => ({ network: blo._network, chainId: utils.evmChainToId(blo._network), blockHash: blo.hash, blockNumber: blo.number, timestamp: utils.blockToTimestamp(blo), ...x, })); }"' | jq .
 	//
@@ -81,7 +31,7 @@ func (d *Daemon) Ensure() error {
 	}
 
 	{
-		res, err := tra.CreateTransformation("test-transformation", cod)
+		res, err := d.tra.CreateTransformation(fmt.Sprintf("%s-transformation", d.env.Environment), cod)
 		if err != nil {
 			tracer.Panic(tracer.Mask(err))
 		}
@@ -98,9 +48,9 @@ func (d *Daemon) Ensure() error {
 	var cpr pipelines.CreatePipelineRequest
 	{
 		cpr = pipelines.CreatePipelineRequest{
-			Name:           "test-pipeline",
-			Transformation: "test-transformation",
-			Filter:         "test-filter",
+			Name:           fmt.Sprintf("%s-pipeline", d.env.Environment),
+			Transformation: fmt.Sprintf("%s-transformation", d.env.Environment),
+			Filter:         fmt.Sprintf("%s-filter", d.env.Environment),
 			FilterKeys:     []string{"from", "to"},
 			Networks:       []string{"ethereum", "base"},
 			Enabled:        true,
@@ -117,7 +67,7 @@ func (d *Daemon) Ensure() error {
 	}
 
 	{
-		res, err := pip.CreatePipeline(cpr)
+		res, err := d.pip.CreatePipeline(cpr)
 		if err != nil {
 			tracer.Panic(tracer.Mask(err))
 		}
@@ -141,7 +91,7 @@ func (d *Daemon) Ensure() error {
 	// }
 
 	// {
-	// 	res, err := pip.BackfillPipeline("test-pipeline", bpr)
+	// 	res, err := d.pip.BackfillPipeline(fmt.Sprintf("%s-pipeline", d.env.Environment), bpr)
 	// 	if err != nil {
 	// 		tracer.Panic(tracer.Mask(err))
 	// 	}

pkg/daemon/worker.go

@@ -1,13 +1,31 @@
 package daemon
 
 import (
-	"github.com/0xSplits/pulsar/pkg/worker/handler/image"
+	"fmt"
+
+	"github.com/0xSplits/pulsar/pkg/worker/handler/pipeline"
+	"github.com/0xSplits/pulsar/pkg/worker/handler/pipeline/accounts"
+	"github.com/0xSplits/pulsar/pkg/worker/handler/pipeline/accounts/client"
 	"github.com/0xSplits/workit/handler"
 	"github.com/0xSplits/workit/registry"
 	"github.com/0xSplits/workit/worker/parallel"
 )
 
 func (d *Daemon) Worker() *parallel.Worker {
+	var cli *client.Client
+	{
+		cli = client.New(client.Config{
+			Url: fmt.Sprintf("https://server.%s.splits.org", d.env.Environment),
+		})
+	}
+
+	var acc *accounts.Accounts
+	{
+		acc = accounts.New(accounts.Config{
+			Cli: cli,
+		})
+	}
+
 	var reg *registry.Registry
 	{
 		reg = registry.New(registry.Config{
@@ -21,7 +39,7 @@ func (d *Daemon) Worker() *parallel.Worker {
 	{
 		par = parallel.New(parallel.Config{
 			Han: []handler.Cooler{
-				image.New(image.Config{Log: d.log}),
+				pipeline.New(pipeline.Config{Acc: acc, Env: d.env, Fil: d.fil, Log: d.log}),
 			},
 			Log: d.log,
 			Reg: reg,

pkg/envvar/envvar.go

@@ -12,6 +12,7 @@ import (
 type Env struct {
 	Environment      string `split_words:"true" required:"true"`
 	IndexingcoApiKey string `split_words:"true" required:"true"`
+	PostgresUrl      string `split_words:"true" required:"true"`
 	WebsocketSecret  string `split_words:"true" required:"true"`
 
 	HttpHost string `split_words:"true" required:"true"`