grassrootseconomics/eth-indexer
7
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: queries

Browse Source
This commit is contained in:
Mohamed Sohail 2024-06-06 14:53:32 +08:00
parent 73d14ec021
commit 7e0fc95918
Signed by: kamikazechaser
GPG Key ID: 7DD45520C01CD85D
7 changed files with 31 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
.env.example Normal file
View File

@ -0,0 +1,5 @@
REMOTE_DB_HOST=
REMOTE_DB_PORT=
REMOTE_DB_NAME=
REMOTE_DB_USER=
REMOTE_DB_PASSWORD=

5
.gitignore vendored Normal file
View File

@ -0,0 +1,5 @@
tracker_db
db
.vscode
.idx
.env

29
internal/store/pg.go
View File

@ -23,6 +23,7 @@ type (
}
Pg struct {
logg *slog.Logger
db *pgxpool.Pool
queries *queries
}
@ -62,6 +63,7 @@ func NewPgStore(o PgOpts) (Store, error) {
o.Logg.Info("migrations ran successfully")
return &Pg{
logg: o.Logg,
db: dbPool,
queries: queries,
}, nil
@ -69,20 +71,21 @@ func NewPgStore(o PgOpts) (Store, error) {
func (pg *Pg) InsertTokenTransfer(ctx context.Context, eventPayload event.Event) error {
return pg.executeTransaction(ctx, func(tx pgx.Tx) error {
var addressExists bool
// var addressExists bool
if err := tx.QueryRow(
ctx,
pg.queries.CheckAddressExists,
eventPayload.Payload["from"].(string),
eventPayload.Payload["to"].(string),
).Scan(&addressExists); err != nil {
return err
}
// if err := tx.QueryRow(
// ctx,
// pg.queries.CheckAddressExists,
// eventPayload.Payload["from"].(string),
// eventPayload.Payload["to"].(string),
// ).Scan(&addressExists); err != nil {
// return err
// }
if !addressExists {
return nil
}
// if !addressExists {
// return nil
// }
// pg.logg.Debug("transfer address exists", "hash", eventPayload.TxHash)
txID, err := pg.insertTx(ctx, tx, eventPayload)
if err != nil {
@ -278,7 +281,7 @@ func runMigrations(ctx context.Context, dbPool *pgxpool.Pool, migrationsPath str
}
}
const migratorTimeout = 5 * time.Second
const migratorTimeout = 15 * time.Second
ctx, cancel := context.WithTimeout(ctx, migratorTimeout)
defer cancel()

5
internal/sub/jetstream.go
View File

@ -55,9 +55,8 @@ func NewJetStreamSub(o JetStreamOpts) (Sub, error) {
}
consumer, err := stream.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{
Durable: o.JetStreamID,
AckPolicy: jetstream.AckExplicitPolicy,
FilterSubject: pullStream,
Durable: o.JetStreamID,
AckPolicy: jetstream.AckExplicitPolicy,
})
if err != nil {
return nil, err

4
migrations/001_indexer_base.sql
View File

@ -4,9 +4,9 @@ CREATE EXTENSION IF NOT EXISTS postgres_fdw;
-- link sarafu_network db
CREATE SCHEMA IF NOT EXISTS sarafu_network;
CREATE SERVER IF NOT EXISTS sarafu_network_remote FOREIGN DATA WRAPPER postgres_fdw OPTIONS
(host '{{.remote_db_host }}', port '{{.remote_db_port }}', dbname '{{.remote_db_name }}');
(host '{{env "REMOTE_DB_HOST" }}', port '{{env "REMOTE_DB_PORT" }}', dbname '{{env "REMOTE_DB_NAME" }}');
CREATE USER MAPPING IF NOT EXISTS FOR postgres SERVER sarafu_network_remote OPTIONS
(user '{{.remote_db_user }}', password '{{.remote_db_password }}');
(user '{{env "REMOTE_DB_USER" }}', password '{{env "REMOTE_DB_PASSWORD" }}');
IMPORT FOREIGN SCHEMA public LIMIT TO (accounts) FROM SERVER sarafu_network_remote INTO sarafu_network;
--

6
migrations/tern.conf
View File

@ -1,6 +0,0 @@
[data]
remote_db_host = {{env "REMOTE_DB_HOST"}}
remote_db_port = {{env "REMOTE_DB_PORT"}}
remote_db_name = {{env "REMOTE_DB_NAME"}}
remote_db_user = {{env "REMOTE_DB_USER"}}
remote_db_password = {{env "REMOTE_DB_PASSWORD"}}

2
queries.sql
View File

@ -104,7 +104,7 @@ INSERT INTO pool_deposit(
INSERT INTO price_index_updates(
tx_id,
token,
exchange_rate,
exchange_rate
) VALUES($1, $2, $3) ON CONFLICT DO NOTHING
--name: address-exists