fix(backend): make analytics migrations compatible with supported PostgreSQL (#1702)

.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.

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");

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

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,

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$;

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'),
   };