diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 5e9955e4..f74054cb 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -120,7 +120,7 @@ jobs: - name: Setup Database run: | npx prisma generate --schema=prisma/schema.prisma - npx prisma db push --accept-data-loss --schema=prisma/schema.prisma + npx prisma migrate deploy --schema=prisma/schema.prisma working-directory: backend env: DATABASE_URL: postgresql://postgres:password@127.0.0.1:5432/flowfi_test @@ -193,7 +193,7 @@ jobs: # `prisma generate` (and rejects the old `--skip-generate` flag), so # only the schema push runs and nothing writes into the root-owned # node_modules tree. - docker compose run --rm -e DATABASE_URL=postgresql://flowfi:flowfi_dev_password@postgres:5432/flowfi backend npx -y prisma db push --accept-data-loss + docker compose run --rm -e DATABASE_URL=postgresql://flowfi:flowfi_dev_password@postgres:5432/flowfi backend npx prisma migrate deploy docker compose up -d backend # Poll /health instead of a fixed sleep so cold-boot / Prisma-warmup # latency does not flake the Backend Docker Image CI. diff --git a/backend/prisma/migrations/20260221132622_init/migration.sql b/backend/prisma/migrations/20260221132622_init/migration.sql index 1dd8f535..529afd99 100644 --- a/backend/prisma/migrations/20260221132622_init/migration.sql +++ b/backend/prisma/migrations/20260221132622_init/migration.sql @@ -3,7 +3,8 @@ CREATE TABLE "User" ( "id" TEXT NOT NULL PRIMARY KEY, "publicKey" TEXT NOT NULL, "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, - "updatedAt" TIMESTAMP(3) NOT NULL + "updatedAt" TIMESTAMP(3) NOT NULL, + CONSTRAINT "User_publicKey_key" UNIQUE ("publicKey") ); -- CreateTable @@ -21,6 +22,7 @@ CREATE TABLE "Stream" ( "isActive" BOOLEAN NOT NULL DEFAULT true, "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, "updatedAt" TIMESTAMP(3) NOT NULL, + CONSTRAINT "Stream_streamId_key" UNIQUE ("streamId"), CONSTRAINT "Stream_sender_fkey" FOREIGN KEY ("sender") REFERENCES "User" ("publicKey") ON DELETE RESTRICT ON UPDATE CASCADE, CONSTRAINT "Stream_recipient_fkey" FOREIGN KEY ("recipient") REFERENCES "User" ("publicKey") ON DELETE RESTRICT ON UPDATE CASCADE ); @@ -40,13 +42,13 @@ CREATE TABLE "StreamEvent" ( ); -- CreateIndex -CREATE UNIQUE INDEX "User_publicKey_key" ON "User"("publicKey"); +CREATE UNIQUE INDEX IF NOT EXISTS "User_publicKey_key" ON "User"("publicKey"); -- CreateIndex CREATE INDEX "User_publicKey_idx" ON "User"("publicKey"); -- CreateIndex -CREATE UNIQUE INDEX "Stream_streamId_key" ON "Stream"("streamId"); +CREATE UNIQUE INDEX IF NOT EXISTS "Stream_streamId_key" ON "Stream"("streamId"); -- CreateIndex CREATE INDEX "Stream_sender_idx" ON "Stream"("sender"); diff --git a/backend/prisma/migrations/20260729170000_timestamp_bigint/migration.sql b/backend/prisma/migrations/20260729170000_timestamp_bigint/migration.sql index e419f40e..fa923eff 100644 --- a/backend/prisma/migrations/20260729170000_timestamp_bigint/migration.sql +++ b/backend/prisma/migrations/20260729170000_timestamp_bigint/migration.sql @@ -1,7 +1,7 @@ -- AlterTable -ALTER TABLE "Stream" ALTER COLUMN "startTime" SET DATA TYPE BIGINT, +ALTER TABLE "Stream" ADD COLUMN IF NOT EXISTS "endTime" BIGINT, +ALTER COLUMN "startTime" SET DATA TYPE BIGINT, ALTER COLUMN "lastUpdateTime" SET DATA TYPE BIGINT, -ALTER COLUMN "endTime" SET DATA TYPE BIGINT, ALTER COLUMN "pausedAt" SET DATA TYPE BIGINT; -- AlterTable diff --git a/backend/prisma/migrations/20260926000000_add_indexer_dead_letter_event/migration.sql b/backend/prisma/migrations/20260926000000_add_indexer_dead_letter_event/migration.sql index 1d42a0ff..b7311529 100644 --- a/backend/prisma/migrations/20260926000000_add_indexer_dead_letter_event/migration.sql +++ b/backend/prisma/migrations/20260926000000_add_indexer_dead_letter_event/migration.sql @@ -1,3 +1,5 @@ +DROP TABLE IF EXISTS "IndexerDeadLetterEvent" CASCADE; + -- CreateTable CREATE TABLE "IndexerDeadLetterEvent" ( "id" TEXT NOT NULL, diff --git a/backend/prisma/migrations/20261001000000_add_timescaledb_analytics/migration.sql b/backend/prisma/migrations/20261001000000_add_timescaledb_analytics/migration.sql index 86185628..d3f12679 100644 --- a/backend/prisma/migrations/20261001000000_add_timescaledb_analytics/migration.sql +++ b/backend/prisma/migrations/20261001000000_add_timescaledb_analytics/migration.sql @@ -1,77 +1,84 @@ --- TimescaleDB analytics backbone (#1480). +-- TimescaleDB analytics backbone (#1480, #1702). -- -- Protocol TVL, streaming velocity and withdrawn totals change with every -- ledger second, so they are snapshotted into a hypertable and pre-aggregated -- with continuous aggregates instead of being recomputed from the raw Stream -- and StreamEvent tables on every request. -- --- Requires the timescaledb extension to be available on the PostgreSQL --- instance (see docker-compose.yml). If the extension is not installed the --- statements below fail; the analytics API falls back to live Prisma --- aggregation in that case, so existing deployments keep working. +-- If the timescaledb extension is available on the PostgreSQL instance, this +-- migration creates the hypertable, continuous aggregates and refresh policies. +-- If the extension is not installed (e.g. stock PostgreSQL in docker-compose or CI), +-- setup is safely skipped and the analytics API falls back to live Prisma aggregation. -CREATE EXTENSION IF NOT EXISTS timescaledb CASCADE; +DO $migration$ +BEGIN + IF EXISTS (SELECT 1 FROM pg_available_extensions WHERE name = 'timescaledb') THEN + EXECUTE 'CREATE EXTENSION IF NOT EXISTS timescaledb CASCADE'; --- Hypertable for continuous stream flow snapshots. One row per --- (token_address, snapshot tick) produced by the indexer's aggregation pass. -CREATE TABLE IF NOT EXISTS stream_flow_snapshots ( - time TIMESTAMPTZ NOT NULL, - token_address TEXT NOT NULL, - active_stream_count INT NOT NULL, - total_locked_amount NUMERIC(38, 0) NOT NULL, - cumulative_streamed_amount NUMERIC(38, 0) NOT NULL, - cumulative_withdrawn_amount NUMERIC(38, 0) NOT NULL, - flow_velocity_per_second NUMERIC(38, 0) NOT NULL -); + -- Hypertable for continuous stream flow snapshots. One row per + -- (token_address, snapshot tick) produced by the indexer's aggregation pass. + EXECUTE 'CREATE TABLE IF NOT EXISTS stream_flow_snapshots ( + time TIMESTAMPTZ NOT NULL, + token_address TEXT NOT NULL, + active_stream_count INT NOT NULL, + total_locked_amount NUMERIC(38, 0) NOT NULL, + cumulative_streamed_amount NUMERIC(38, 0) NOT NULL, + cumulative_withdrawn_amount NUMERIC(38, 0) NOT NULL, + flow_velocity_per_second NUMERIC(38, 0) NOT NULL + )'; -SELECT create_hypertable('stream_flow_snapshots', 'time', if_not_exists => TRUE); + EXECUTE 'SELECT create_hypertable(''stream_flow_snapshots'', ''time'', if_not_exists => TRUE)'; -CREATE INDEX IF NOT EXISTS idx_stream_flow_snapshots_token_time - ON stream_flow_snapshots (token_address, time DESC); + EXECUTE 'CREATE INDEX IF NOT EXISTS idx_stream_flow_snapshots_token_time + ON stream_flow_snapshots (token_address, time DESC)'; --- Hourly continuous aggregate: average TVL, summed velocity and peak stream --- count per token, refreshable incrementally by TimescaleDB policies. -CREATE MATERIALIZED VIEW IF NOT EXISTS hourly_protocol_metrics -WITH (timescaledb.continuous) AS -SELECT time_bucket('1 hour', time) AS bucket, - token_address, - AVG(total_locked_amount) AS avg_tvl, - SUM(flow_velocity_per_second) AS aggregate_velocity, - MAX(active_stream_count) AS peak_streams -FROM stream_flow_snapshots -GROUP BY bucket, token_address -WITH NO DATA; + -- Hourly continuous aggregate: average TVL, summed velocity and peak stream + -- count per token, refreshable incrementally by TimescaleDB policies. + EXECUTE 'CREATE MATERIALIZED VIEW IF NOT EXISTS hourly_protocol_metrics + WITH (timescaledb.continuous) AS + SELECT time_bucket(''1 hour'', time) AS bucket, + token_address, + AVG(total_locked_amount) AS avg_tvl, + SUM(flow_velocity_per_second) AS aggregate_velocity, + MAX(active_stream_count) AS peak_streams + FROM stream_flow_snapshots + GROUP BY bucket, token_address + WITH NO DATA'; --- Daily continuous aggregate for 30d/90d/1y historical charts. -CREATE MATERIALIZED VIEW IF NOT EXISTS daily_protocol_metrics -WITH (timescaledb.continuous) AS -SELECT time_bucket('1 day', time) AS bucket, - token_address, - AVG(total_locked_amount) AS avg_tvl, - SUM(flow_velocity_per_second) AS aggregate_velocity, - MAX(active_stream_count) AS peak_streams, - MAX(cumulative_withdrawn_amount) AS cumulative_withdrawn -FROM stream_flow_snapshots -GROUP BY bucket, token_address -WITH NO DATA; + -- Daily continuous aggregate for 30d/90d/1y historical charts. + EXECUTE 'CREATE MATERIALIZED VIEW IF NOT EXISTS daily_protocol_metrics + WITH (timescaledb.continuous) AS + SELECT time_bucket(''1 day'', time) AS bucket, + token_address, + AVG(total_locked_amount) AS avg_tvl, + SUM(flow_velocity_per_second) AS aggregate_velocity, + MAX(active_stream_count) AS peak_streams, + MAX(cumulative_withdrawn_amount) AS cumulative_withdrawn + FROM stream_flow_snapshots + GROUP BY bucket, token_address + WITH NO DATA'; --- Refresh policies: hourly aggregate refreshes shortly after each hour ends, --- daily aggregate in the small hours. Both retain their full window. -SELECT add_continuous_aggregate_policy('hourly_protocol_metrics', - start_offset => INTERVAL '3 days', - end_offset => INTERVAL '1 hour', - schedule_interval => INTERVAL '1 hour'); + -- Refresh policies: hourly aggregate refreshes shortly after each hour ends, + -- daily aggregate in the small hours. Both retain their full window. + EXECUTE 'SELECT add_continuous_aggregate_policy(''hourly_protocol_metrics'', + start_offset => INTERVAL ''3 days'', + end_offset => INTERVAL ''1 hour'', + schedule_interval => INTERVAL ''1 hour'')'; -SELECT add_continuous_aggregate_policy('daily_protocol_metrics', - start_offset => INTERVAL '40 days', - end_offset => INTERVAL '1 day', - schedule_interval => INTERVAL '1 day'); + EXECUTE 'SELECT add_continuous_aggregate_policy(''daily_protocol_metrics'', + start_offset => INTERVAL ''40 days'', + end_offset => INTERVAL ''1 day'', + schedule_interval => INTERVAL ''1 day'')'; --- Compress raw snapshots older than 14 days to keep the hypertable small --- while retaining every data point for the daily rollups. -ALTER TABLE stream_flow_snapshots SET ( - timescaledb.compress, - timescaledb.compress_segmentby = 'token_address' -); + -- Compress raw snapshots older than 14 days to keep the hypertable small + -- while retaining every data point for the daily rollups. + EXECUTE 'ALTER TABLE stream_flow_snapshots SET ( + timescaledb.compress, + timescaledb.compress_segmentby = ''token_address'' + )'; -SELECT add_compression_policy('stream_flow_snapshots', INTERVAL '14 days'); + EXECUTE 'SELECT add_compression_policy(''stream_flow_snapshots'', INTERVAL ''14 days'')'; + ELSE + RAISE NOTICE 'TimescaleDB extension is not available; skipping analytics continuous aggregates setup. Live analytics fallback will be used.'; + END IF; +END $migration$; diff --git a/backend/src/services/indexerService.ts b/backend/src/services/indexerService.ts index abb7cda3..1eca8781 100644 --- a/backend/src/services/indexerService.ts +++ b/backend/src/services/indexerService.ts @@ -240,7 +240,7 @@ export function serializeDeadLetterPayload(event: rpc.Api.EventResponse): string ledgerClosedAt: event.ledgerClosedAt, transactionIndex: event.transactionIndex, operationIndex: event.operationIndex, - inSuccessfulContractCall: event.inSuccessfulContractCall, + inSuccessfulContractCall: event.inSuccessfulContractCall ?? false, topic: event.topic.map((scval) => Buffer.from(scval.toXDR()).toString('base64')), value: Buffer.from(event.value.toXDR()).toString('base64'), }; diff --git a/frontend/src/lib/api-types.generated.ts b/frontend/src/lib/api-types.generated.ts index bb0db700..02d2c791 100644 --- a/frontend/src/lib/api-types.generated.ts +++ b/frontend/src/lib/api-types.generated.ts @@ -2577,6 +2577,197 @@ export interface paths { patch?: never; trace?: never; }; + "/v1/analytics/tvl": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + /** + * Current protocol TVL partitioned by token asset + * @description Read-only protocol statistics. No authentication is required because + * every payload here is public (the same numbers a block explorer + * would show). Subject to the API's global rate limiter. + */ + get: { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + requestBody?: never; + responses: { + /** @description TVL snapshot */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": { + /** @example true */ + success: boolean; + /** @description Current TVL snapshot partitioned by token asset */ + data: Record; + }; + }; + }; + /** @description Too many requests */ + 429: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + /** @description TVL snapshot unavailable */ + 500: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + }; + }; + put?: never; + post?: never; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; + "/v1/analytics/historical": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + /** + * Pre-aggregated time-series points for frontend charts + * @description Historical TVL and volume series at a fixed interval. No + * authentication is required; subject to the API's global rate limiter. + */ + get: { + parameters: { + query?: { + /** @description Lookback window */ + period?: "7d" | "30d" | "90d" | "1y"; + /** @description Granularity of each point */ + interval?: "1h" | "1d"; + }; + header?: never; + path?: never; + cookie?: never; + }; + requestBody?: never; + responses: { + /** @description Time-series points */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": { + /** @example true */ + success: boolean; + /** @description Aggregated time-series points for the requested period and interval */ + data: unknown[]; + }; + }; + }; + /** @description Invalid period or interval */ + 400: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + /** @description Too many requests */ + 429: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + /** @description Historical analytics unavailable */ + 500: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + }; + }; + put?: never; + post?: never; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; + "/v1/analytics/defillama": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + /** + * DefiLlama adapter payload + * @description TVL and activity numbers in the shape DefiLlama's adapter format + * expects, so the protocol can be listed without a separate adapter. + * No authentication is required; subject to the global rate limiter. + */ + get: { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + requestBody?: never; + responses: { + /** @description DefiLlama adapter payload */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": { + /** @example true */ + success: boolean; + /** @description DefiLlama-shaped adapter payload */ + data: Record; + }; + }; + }; + /** @description Too many requests */ + 429: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + /** @description DefiLlama adapter unavailable */ + 500: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + }; + }; + put?: never; + post?: never; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; "/v1/admin/metrics": { parameters: { query?: never;