feat(indexer): postgresql connection pooling and graphql query comple…

indexer/knexfile.js

@@ -0,0 +1,59 @@
+'use strict';
+
+/**
+ * Knex configuration for SoroTask indexer PostgreSQL & TimescaleDB database.
+ * Supports connection pooling and migration management.
+ */
+
+const path = require('path');
+
+const DATABASE_URL =
+  process.env.DATABASE_URL ||
+  process.env.POSTGRES_URL ||
+  'postgresql://postgres:postgres@localhost:5432/sorotask_indexer';
+
+module.exports = {
+  development: {
+    client: 'pg',
+    connection: DATABASE_URL,
+    pool: {
+      min: parseInt(process.env.DATABASE_POOL_MIN || '2', 10),
+      max: parseInt(process.env.DATABASE_WRITE_POOL_MAX || '20', 10),
+      idleTimeoutMillis: parseInt(process.env.DATABASE_IDLE_TIMEOUT_MS || '30000', 10),
+      acquireTimeoutMillis: parseInt(process.env.DATABASE_CONN_TIMEOUT_MS || '10000', 10),
+    },
+    migrations: {
+      directory: path.join(__dirname, 'migrations'),
+      tableName: 'knex_migrations',
+    },
+  },
+
+  test: {
+    client: 'pg',
+    connection: process.env.TEST_DATABASE_URL || DATABASE_URL,
+    pool: {
+      min: 1,
+      max: 10,
+      idleTimeoutMillis: 10000,
+    },
+    migrations: {
+      directory: path.join(__dirname, 'migrations'),
+      tableName: 'knex_migrations',
+    },
+  },
+
+  production: {
+    client: 'pg',
+    connection: DATABASE_URL,
+    pool: {
+      min: parseInt(process.env.DATABASE_POOL_MIN || '5', 10),
+      max: parseInt(process.env.DATABASE_WRITE_POOL_MAX || '50', 10),
+      idleTimeoutMillis: parseInt(process.env.DATABASE_IDLE_TIMEOUT_MS || '30000', 10),
+      acquireTimeoutMillis: parseInt(process.env.DATABASE_CONN_TIMEOUT_MS || '10000', 10),
+    },
+    migrations: {
+      directory: path.join(__dirname, 'migrations'),
+      tableName: 'knex_migrations',
+    },
+  },
+};

indexer/migrations/002_timescaledb_raw_events_retention.down.sql

@@ -21,4 +21,4 @@ ALTER TABLE raw_events DROP CONSTRAINT IF EXISTS raw_events_dedup_key;
 ALTER TABLE raw_events DROP COLUMN IF EXISTS ledger_timestamp;
 
 -- Restore the original table name used by 001_initial_schema.
-ALTER TABLE IF EXISTS raw_events RENAME TO events;
+ALTER TABLE IF EXISTS raw_events RENAME TO events;

indexer/migrations/004_timescaledb_executions_hypertable.down.sql

@@ -0,0 +1,4 @@
+-- Down migration: 004_timescaledb_executions_hypertable
+-- Reverses TimescaleDB hypertable compression on executions.
+
+SELECT remove_compression_policy('executions', if_exists => TRUE);

indexer/migrations/004_timescaledb_executions_hypertable.sql

@@ -0,0 +1,33 @@
+-- Migration: 004_timescaledb_executions_hypertable
+-- Creates TimescaleDB hypertable for execution logs (executions table)
+-- with compression and retention policies for high-throughput execution tracking.
+
+CREATE EXTENSION IF NOT EXISTS timescaledb;
+
+-- TimescaleDB hypertables require primary key/unique constraints to include partitioning column
+ALTER TABLE executions DROP CONSTRAINT IF EXISTS executions_pkey;
+ALTER TABLE executions ADD CONSTRAINT executions_pkey PRIMARY KEY (id, executed_at);
+
+SELECT create_hypertable(
+    'executions',
+    'executed_at',
+    chunk_time_interval => INTERVAL '7 days',
+    migrate_data => TRUE,
+    if_not_exists => TRUE
+);
+
+ALTER TABLE executions SET (
+    timescaledb.compress,
+    timescaledb.compress_segmentby = 'task_id, keeper_address, status',
+    timescaledb.compress_orderby = 'executed_at DESC, id DESC'
+);
+
+SELECT add_compression_policy(
+    'executions',
+    INTERVAL '14 days',
+    if_not_exists => TRUE
+);
+
+INSERT INTO schema_migrations (version)
+VALUES ('004_timescaledb_executions_hypertable')
+ON CONFLICT (version) DO NOTHING;