Store FOMO WebSocket Events in Postgres

A FOMO WebSocket Postgres pipeline needs one property above all: receiving the same alert twice must not create two database events. Reconnect replays are useful for filling gaps, but they make process-local deduplication insufficient. Put the uniqueness rule in Postgres and retain the raw message alongside the fields you query.

This guide focuses on persistence for the app-feed stream at wss://api.fomoapi.io/ws/alerts. Use the production WebSocket client for heartbeat monitoring, bounded reconnects, and filter management.

Install the Node dependencies

npm install ws pg
export FOMO_API_KEY="your-key"
export DATABASE_URL="postgresql://user:pass@host:5432/app"

Keep both values in server-side secrets. A WebSocket URL contains the API key as a query parameter, so do not print the full URL in logs or exception messages.

Create an append-only event table

Keep a compact set of typed columns for filters and joins, plus raw for forward compatibility. The feed can add fields without forcing an emergency schema migration.

create table if not exists fomo_alert_events (
  id bigint generated always as identity primary key,
  dedupe_key text not null unique,
  event_id text,
  stream_id text,
  user_id text,
  trade_id text,
  alert_type text not null,
  source text,
  trader text,
  token text,
  token_address text,
  chain_id bigint,
  chain text,
  event_ts bigint,
  replay boolean not null default false,
  trade_usd numeric,
  position_value_usd numeric,
  realized_pnl_usd numeric,
  usd_value numeric,
  raw jsonb not null,
  received_at timestamptz not null default now()
);

create index if not exists fomo_alert_events_trader_ts
  on fomo_alert_events (trader, event_ts desc);

create index if not exists fomo_alert_events_token_ts
  on fomo_alert_events (token_address, event_ts desc)
  where token_address is not null;

The eventId field is the preferred dedupe identity on feed events. Some push-derived events do not carry FOMO object IDs, so the ingestion code falls back to the stream message id. If neither exists, reject the event or compute a carefully versioned content hash; do not silently create a random key that defeats deduplication.

Insert with a database uniqueness rule

Use placeholders for every value. ON CONFLICT DO NOTHING turns replayed delivery into a harmless no-op even after a restart or when two ingestion replicas overlap.

import pg from "pg";

const { Pool } = pg;
const pool = new Pool({ connectionString: process.env.DATABASE_URL });

const INSERT_ALERT = [
  "insert into fomo_alert_events (",
  "  dedupe_key, event_id, stream_id, user_id, trade_id,",
  "  alert_type, source, trader, token, token_address,",
  "  chain_id, chain, event_ts, replay, trade_usd,",
  "  position_value_usd, realized_pnl_usd, usd_value, raw",
  ") values (",
  "  $1, $2, $3, $4, $5, $6, $7, $8, $9, $10,",
  "  $11, $12, $13, $14, $15, $16, $17, $18, $19::jsonb",
  ") on conflict (dedupe_key) do nothing",
].join("\n");

async function persistAlert(message) {
  const stableId = message.eventId || message.id;
  if (!stableId) throw new Error("alert has no stable dedupe identity");

  const dedupeKey = message.eventId
    ? "event:" + message.eventId
    : "stream:" + message.source + ":" + message.id;

  const values = [
    dedupeKey,
    message.eventId ?? null,
    message.id ?? null,
    message.userId ?? null,
    message.tradeId ?? null,
    message.alertType,
    message.source ?? null,
    message.trader ?? null,
    message.token ?? null,
    message.tokenAddress ?? null,
    message.chainId ?? null,
    message.chain ?? null,
    message.ts ?? null,
    message.replay === true,
    message.tradeUsd ?? null,
    message.positionValueUsd ?? null,
    message.realizedPnlUsd ?? null,
    message.usdValue ?? null,
    JSON.stringify(message),
  ];

  const result = await pool.query(INSERT_ALERT, values);
  return result.rowCount === 1;
}

Do not treat usdValue as a universal fill size. Its meaning varies by alert type. The stream provides more specific fields where available: tradeUsd, positionValueUsd, and realizedPnlUsd. Retaining all of them prevents later analytics from assigning the wrong meaning to one number.

Connect ingestion to the alerts stream

Serialize writes within this small example so message handlers do not create an unbounded set of promises. Higher-throughput systems can use a bounded queue and a fixed worker count.

import WebSocket from "ws";

const apiKey = process.env.FOMO_API_KEY;
if (!apiKey) throw new Error("FOMO_API_KEY is required");
if (!process.env.DATABASE_URL) throw new Error("DATABASE_URL is required");

const url = new URL("wss://api.fomoapi.io/ws/alerts");
url.searchParams.set("key", apiKey);

const socket = new WebSocket(url);
let writes = Promise.resolve();

socket.on("message", (bytes) => {
  let message;
  try {
    message = JSON.parse(bytes.toString());
  } catch {
    console.warn("ignored malformed WebSocket frame");
    return;
  }

  if (message.type !== "alert") return;

  writes = writes
    .then(() => persistAlert(message))
    .catch((error) => {
      console.error("alert persistence failed", {
        eventId: message.eventId ?? null,
        error: error.message,
      });
    });
});

socket.on("open", () => console.log("FOMO alert stream connected"));
socket.on("error", (error) => {
  console.error("FOMO alert stream error", error.message);
});

This is deliberately not a complete reconnect client. Production code should wait for a valid welcome frame, monitor application heartbeats, add jittered backoff, and process replay messages. Database uniqueness makes those replays safe rather than something to suppress at the socket.

Recover gaps without changing the data model

After a disconnect, the stream replays recent matching alerts. For a suspected longer gap, GET /v2/alerts can retrieve available recent activity using a since timestamp. Pass each recovered alert through the same persistAlert function. The unique key makes live, replay, and REST recovery converge on one row.

REST history is bounded, so record operational state such as:

• the newest stored event timestamp;
• the last valid WebSocket heartbeat;
• reconnect attempts and welcome delay;
• inserted and duplicate counts;
• queue depth and database write failures.

If REST recovery returns 429 or a transient 5xx, apply the bounded policy in the error-handling guide. Never let recovery requests grow without a retry or time budget.

Query by stable identities

trader is useful for display and filtering, but a handle can change. Prefer user_id for durable trader joins and trade_id for joining an alert to its trade detail. Prefer token_address plus chain identity over a ticker symbol because symbols can collide.

select alert_type, token, chain, event_ts, trade_usd, replay
from fomo_alert_events
where user_id = $1
order by event_ts desc
limit 100;

Add retention only after deciding what the table is for. A short alert cache, an audit log, and an analytics warehouse have different retention and privacy requirements. If rows are deleted, keep that job explicit and observable.

The durable pattern is simple: one stable dedupe key, one append-only insert path, raw events for compatibility, and bounded recovery through the same writer.