Skip to content

Refactoring and Improvements #128

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mschead merged 4 commits into from
Dec 19, 2024
Merged

Refactoring and Improvements #128

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
925cf0b
fix: bugs found by Albert
mschead Dec 17, 2024
a51b4ae
fix: removed unused code on old graphql api
mschead Dec 18, 2024
e3e4d25
refat: removed unused files
mschead Dec 18, 2024
124b6d2
refat: improved streaming
mschead Dec 19, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 6 additions & 16 deletions indexer/.env.template
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,24 +1,14 @@
AWS_S3_REGION="us-east-1"
AWS_S3_BUCKET_NAME="kadena-indexer-data-002"
AWS_ACCESS_KEY_ID="YOUR_ACCESS KEY_ID"
AWS_SECRET_ACCESS_KEY="YOUR_SECRET_ACCESS_KEY"
# only use if you wanna run s3 in your local machine through localstack
AWS_S3_ENDPOINT_LOCAL_STACK=http://localhost:4566

NODE_API_URL=https://api.chainweb.com
SYNC_BASE_URL="https://api.chainweb.com/chainweb/0.0"

SYNC_MIN_HEIGHT=0
SYNC_FETCH_INTERVAL_IN_BLOCKS=50
SYNC_TIME_BETWEEN_REQUESTS_IN_MS=5000
SYNC_ATTEMPTS_MAX_RETRY=10
SYNC_ATTEMPTS_INTERVAL_IN_MS=2000
SYNC_FETCH_INTERVAL_IN_BLOCKS=100
SYNC_NETWORK="mainnet01"
KADENA_GRAPHQL_API_URL=localhost
KADENA_GRAPHQL_API_PORT=3001

DB_USERNAME="postgres"
DB_PASSWORD="YOUR_DB_PASSWORD"
DB_NAME="indexer"
DB_HOST="YOUR_DB_HOST"

RUN_GRAPHQL_ON_START=true
RUN_STREAMING_ON_START=true
RUN_MISSING_BLOCKS_ON_START=false
DB_SSL_ENABLED=false
DB_HOST="YOUR_DB_HOST"
18 changes: 18 additions & 0 deletions indexer/migrations/20241213221154-add-comments-to-avoid-name-clashing-in-postgraphile.js
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,18 @@
"use strict";

/** @type {import('sequelize-cli').Migration} */
module.exports = {
async up(queryInterface) {
await queryInterface.sequelize.query(`
COMMENT ON TABLE "public"."Guards" IS E'@name GuardsPg';
COMMENT ON TABLE "public"."Signers" IS E'@name SignersPg';
`);
},

async down(queryInterface) {
await queryInterface.sequelize.query(`
COMMENT ON TABLE "public"."Guards" IS NULL;
COMMENT ON TABLE "public"."Signers" IS NULL;
`);
},
};
10 changes: 2 additions & 8 deletions indexer/package.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,18 +64,12 @@
"graphql:generate-types": "npx graphql-codegen",
"dev:database": "ts-node src/index.ts --database",
"dev:streaming": "ts-node src/index.ts --streaming",
"dev:backfill": "ts-node src/index.ts --backfill",
"dev:missing": "ts-node src/index.ts --missing",
"dev:headers": "ts-node src/index.ts --headers",
"dev:payloads": "ts-node src/index.ts --payloads",
"dev:graphql": "ts-node src/index.ts --graphql",
"dev:graphqlserver": "nodemon src/index.ts --graphqlServer",
"dev:run": "ts-node src/index.ts --run",
"dev:old-graphql": "ts-node src/index.ts --oldGraphql",
"dev:graphql": "nodemon src/index.ts --graphql",
"prod:start": "docker-compose up --build indexer && docker-compose logs -f indexer",
"prod:streaming": "node dist/index.js --streaming",
"prod:backfill": "node dist/index.js --backfill",
"test": "NODE_ENV=test mocha -r ts-node/register 'tests/**/*.test.ts'",
"start-localstack": "docker run --rm -p 4566:4566 -p 4571:4571 localstack/localstack",
"create:database": "ts-node src/index.ts --database && yarn migrate:up",
"migrate:up": "dotenv -e .env npx sequelize-cli db:migrate",
"migrate:down": "dotenv -e .env npx sequelize-cli db:migrate:undo"
Expand Down
274 changes: 137 additions & 137 deletions indexer/src/config/init.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -148,75 +148,75 @@ export async function initializeDatabase(noTrigger = true): Promise<void> {
EXECUTE FUNCTION update_balances();
`);

console.log('Sync public."Balances"...');

// Create the balances table
await sequelize.query(`
CREATE TABLE IF NOT EXISTS public."Balances" (
id serial4 NOT NULL,
account varchar(255) NOT NULL,
"chainId" int4 NOT NULL,
balance numeric(50) DEFAULT 0 NOT NULL,
"module" varchar(255) NOT NULL,
qualname varchar(255) NOT NULL,
"tokenId" varchar(255) NULL,
network varchar(255) NOT NULL,
"hasTokenId" boolean DEFAULT false NOT NULL,
"contractId" int4 NULL,
"createdAt" timestamptz NOT NULL,
"updatedAt" timestamptz NOT NULL,
CONSTRAINT "Balances_pkey" PRIMARY KEY (id)
);
DO $$
BEGIN
IF NOT EXISTS (
SELECT 1
FROM pg_indexes
WHERE schemaname = 'public'
AND tablename = 'Balances'
AND indexname = 'balance_unique_constraint'
) THEN
CREATE UNIQUE INDEX balance_unique_constraint ON public."Balances" USING btree (network, "chainId", account, qualname, "tokenId");
END IF;
END $$;
`);
// console.log('Sync public."Balances"...');

// // Create the balances table
// await sequelize.query(`
// CREATE TABLE IF NOT EXISTS public."Balances" (
// id serial4 NOT NULL,
// account varchar(255) NOT NULL,
// "chainId" int4 NOT NULL,
// balance numeric(50) DEFAULT 0 NOT NULL,
// "module" varchar(255) NOT NULL,
// qualname varchar(255) NOT NULL,
// "tokenId" varchar(255) NULL,
// network varchar(255) NOT NULL,
// "hasTokenId" boolean DEFAULT false NOT NULL,
// "contractId" int4 NULL,
// "createdAt" timestamptz NOT NULL,
// "updatedAt" timestamptz NOT NULL,
// CONSTRAINT "Balances_pkey" PRIMARY KEY (id)
// );

// DO $$
// BEGIN
// IF NOT EXISTS (
// SELECT 1
// FROM pg_indexes
// WHERE schemaname = 'public'
// AND tablename = 'Balances'
// AND indexname = 'balance_unique_constraint'
// ) THEN
// CREATE UNIQUE INDEX balance_unique_constraint ON public."Balances" USING btree (network, "chainId", account, qualname, "tokenId");
// END IF;
// END $$;
// `);

// --------------------------------
// Missing blocks
// --------------------------------

console.log("Sync missing_block_ranges...");

// Create missing blocks view
await sequelize.query(`
CREATE OR REPLACE VIEW missing_block_ranges AS
WITH missing_ranges AS (
SELECT DISTINCT
"chainId",
"chainwebVersion",
height + 1 AS missing_start,
next_height - 1 AS missing_end
FROM (
SELECT
"chainId",
"chainwebVersion",
height,
LEAD(height) OVER (PARTITION BY "chainId", "chainwebVersion" ORDER BY height) AS next_height
FROM "Blocks"
) AS t
WHERE next_height IS NOT NULL AND next_height <> height + 1
)
SELECT DISTINCT
"chainId",
"chainwebVersion",
missing_start AS from_height,
missing_end AS to_height,
(missing_end - missing_start) AS diff
FROM missing_ranges
where (missing_end - missing_start) >= 0
ORDER BY "chainId", "chainwebVersion", missing_start ASC;
`);
// console.log("Sync missing_block_ranges...");

// // Create missing blocks view
// await sequelize.query(`
// CREATE OR REPLACE VIEW missing_block_ranges AS
// WITH missing_ranges AS (
// SELECT DISTINCT
// "chainId",
// "chainwebVersion",
// height + 1 AS missing_start,
// next_height - 1 AS missing_end
// FROM (
// SELECT
// "chainId",
// "chainwebVersion",
// height,
// LEAD(height) OVER (PARTITION BY "chainId", "chainwebVersion" ORDER BY height) AS next_height
// FROM "Blocks"
// ) AS t
// WHERE next_height IS NOT NULL AND next_height <> height + 1
// )
// SELECT DISTINCT
// "chainId",
// "chainwebVersion",
// missing_start AS from_height,
// missing_end AS to_height,
// (missing_end - missing_start) AS diff
// FROM missing_ranges
// where (missing_end - missing_start) >= 0
// ORDER BY "chainId", "chainwebVersion", missing_start ASC;
// `);

// --------------------------------
// Orphan blocks
Expand Down Expand Up @@ -268,72 +268,72 @@ export async function initializeDatabase(noTrigger = true): Promise<void> {
$function$
;`);

console.log("Sync public.check_backward_orphans()...");

// Create the check backward orphans function
await sequelize.query(`
CREATE OR REPLACE FUNCTION public.check_backward_orphans()
RETURNS trigger
LANGUAGE plpgsql
AS $function$
DECLARE
recent_blocks RECORD;
previous_block RECORD;
first_block RECORD;
block_count INT := 0;
depth CONSTANT INT := 10; -- Default the depth constant
buffer CONSTANT INT := 5; -- Number of heights to buffer, because some blocks can arrive out of order
BEGIN
PERFORM pg_advisory_xact_lock(hashtext(NEW."chainId"::text || NEW."chainwebVersion"::text));
-- Check the last 'depth' blocks
FOR recent_blocks IN
SELECT * FROM public."Blocks"
WHERE height BETWEEN (NEW.height - buffer - depth) AND (NEW.height - buffer - 1)
AND "chainId" = NEW."chainId"
AND "chainwebVersion" = NEW."chainwebVersion"
AND COALESCE(canonical, TRUE)
ORDER BY height ASC
FOR NO KEY UPDATE
LOOP
-- Set the first block
IF block_count = 0 THEN
first_block := recent_blocks;
END IF;
IF previous_block IS NULL THEN
ELSE
-- Check for non-duplicated block
IF previous_block.height = recent_blocks.height
AND (recent_blocks.canonical = FALSE OR recent_blocks.canonical IS NULL) THEN
PERFORM check_canonical(first_block.hash, recent_blocks.height, recent_blocks."chainId", recent_blocks."chainwebVersion", depth);
ELSE
UPDATE public."Blocks"
SET canonical = TRUE
WHERE hash = recent_blocks.hash
AND "chainId" = NEW."chainId"
AND "chainwebVersion" = NEW."chainwebVersion";
END IF;
END IF;
-- Check for gaps
IF recent_blocks.height <> (NEW.height - buffer) - block_count - 1 THEN
-- If there are gaps, do not change canonical status
RETURN NEW;
END IF;
previous_block := recent_blocks;
block_count := block_count + 1;
END LOOP;
IF previous_block IS NULL THEN
RETURN NEW;
END IF;
RETURN NEW;
END;
$function$
;`);
// console.log("Sync public.check_backward_orphans()...");

// // Create the check backward orphans function
// await sequelize.query(`
// CREATE OR REPLACE FUNCTION public.check_backward_orphans()
// RETURNS trigger
// LANGUAGE plpgsql
// AS $function$
// DECLARE
// recent_blocks RECORD;
// previous_block RECORD;
// first_block RECORD;
// block_count INT := 0;
// depth CONSTANT INT := 10; -- Default the depth constant
// buffer CONSTANT INT := 5; -- Number of heights to buffer, because some blocks can arrive out of order
// BEGIN
// PERFORM pg_advisory_xact_lock(hashtext(NEW."chainId"::text || NEW."chainwebVersion"::text));

// -- Check the last 'depth' blocks
// FOR recent_blocks IN
// SELECT * FROM public."Blocks"
// WHERE height BETWEEN (NEW.height - buffer - depth) AND (NEW.height - buffer - 1)
// AND "chainId" = NEW."chainId"
// AND "chainwebVersion" = NEW."chainwebVersion"
// AND COALESCE(canonical, TRUE)
// ORDER BY height ASC
// FOR NO KEY UPDATE
// LOOP
// -- Set the first block
// IF block_count = 0 THEN
// first_block := recent_blocks;
// END IF;

// IF previous_block IS NULL THEN
// ELSE
// -- Check for non-duplicated block
// IF previous_block.height = recent_blocks.height
// AND (recent_blocks.canonical = FALSE OR recent_blocks.canonical IS NULL) THEN
// PERFORM check_canonical(first_block.hash, recent_blocks.height, recent_blocks."chainId", recent_blocks."chainwebVersion", depth);
// ELSE
// UPDATE public."Blocks"
// SET canonical = TRUE
// WHERE hash = recent_blocks.hash
// AND "chainId" = NEW."chainId"
// AND "chainwebVersion" = NEW."chainwebVersion";
// END IF;
// END IF;

// -- Check for gaps
// IF recent_blocks.height <> (NEW.height - buffer) - block_count - 1 THEN
// -- If there are gaps, do not change canonical status
// RETURN NEW;
// END IF;

// previous_block := recent_blocks;
// block_count := block_count + 1;
// END LOOP;

// IF previous_block IS NULL THEN
// RETURN NEW;
// END IF;

// RETURN NEW;
// END;
// $function$
// ;`);

console.log("Sync public.check_upward_orphans()...");

Expand Down Expand Up @@ -455,14 +455,14 @@ export async function initializeDatabase(noTrigger = true): Promise<void> {
FOR EACH ROW
EXECUTE FUNCTION transactions_propagate_canonical_function();`);

console.log("Sync check_orphan_blocks_backward...");
// console.log("Sync check_orphan_blocks_backward...");

// Create orphan blocks trigger
await sequelize.query(`
CREATE OR REPLACE TRIGGER check_orphan_blocks_backward
AFTER INSERT ON public."Blocks"
FOR EACH ROW
EXECUTE FUNCTION check_backward_orphans();`);
// // Create orphan blocks trigger
// await sequelize.query(`
// CREATE OR REPLACE TRIGGER check_orphan_blocks_backward
// AFTER INSERT ON public."Blocks"
// FOR EACH ROW
// EXECUTE FUNCTION check_backward_orphans();`);

console.log("Sync check_orphan_blocks_upward...");

Expand Down
Loading
Loading