polymarket-analyst_github/api/_db.js

import { neon } from "@neondatabase/serverless";

let sqlClient;
let schemaReady;

export function databaseUrl() {
  return process.env.DATABASE_URL || process.env.NEON_DATABASE_URL || "";
}

export function hasDatabase() {
  return Boolean(databaseUrl());
}

export function sql() {
  if (!sqlClient) {
    const url = databaseUrl();
    if (!url) throw new Error("Database is not configured");
    sqlClient = neon(url);
  }
  return sqlClient;
}

export async function ensureSchema() {
  if (schemaReady) return;
  const db = sql();

  await db`
    create table if not exists provider_events (
      id bigserial primary key,
      provider text not null,
      event_type text not null,
      external_id text,
      signature_valid boolean not null default false,
      payload jsonb not null,
      headers jsonb not null default '{}'::jsonb,
      received_at timestamptz not null default now()
    )
  `;

  await db`
    create unique index if not exists provider_events_provider_external_id_idx
      on provider_events (provider, external_id)
      where external_id is not null
  `;

  await db`
    create table if not exists readiness_audit (
      id bigserial primary key,
      event_type text not null,
      detail jsonb not null,
      created_at timestamptz not null default now()
    )
  `;

  await db`
    create table if not exists user_consents (
      id bigserial primary key,
      user_id text not null,
      terms_version text not null,
      privacy_version text not null,
      risk_disclosure_version text not null,
      jurisdiction text,
      payload jsonb not null default '{}'::jsonb,
      accepted_at timestamptz not null default now()
    )
  `;

  await db`
    create table if not exists risk_profiles (
      user_id text primary key,
      wallet_address text,
      jurisdiction text,
      deposit_limit numeric not null default 0,
      withdraw_reserve numeric not null default 0,
      agent_permissions jsonb not null default '{}'::jsonb,
      updated_at timestamptz not null default now()
    )
  `;

  await db`
    create table if not exists trade_tickets (
      id bigserial primary key,
      user_id text not null,
      agent_id text not null,
      market_id text not null,
      question text,
      market_url text,
      side text not null,
      max_amount numeric not null,
      limit_price numeric,
      rationale text,
      status text not null default 'staged',
      instructions jsonb not null default '{}'::jsonb,
      created_at timestamptz not null default now(),
      updated_at timestamptz not null default now()
    )
  `;

  await db`
    create index if not exists trade_tickets_user_status_idx
      on trade_tickets (user_id, status, created_at desc)
  `;

  await db`
    create table if not exists real_fills (
      id bigserial primary key,
      user_id text not null,
      ticket_id bigint,
      agent_id text,
      market_id text not null,
      question text,
      market_url text,
      side text not null,
      action text not null,
      shares numeric not null,
      price numeric not null,
      fees numeric not null default 0,
      tx_note text,
      filled_at timestamptz not null default now(),
      created_at timestamptz not null default now()
    )
  `;

  await db`
    create index if not exists real_fills_user_market_idx
      on real_fills (user_id, market_id, side, filled_at desc)
  `;

  await db`
    create table if not exists real_positions (
      id bigserial primary key,
      user_id text not null,
      agent_id text,
      market_id text not null,
      question text,
      market_url text,
      side text not null,
      shares numeric not null default 0,
      avg_price numeric not null default 0,
      cost_basis numeric not null default 0,
      realized_pnl numeric not null default 0,
      current_price numeric,
      updated_at timestamptz not null default now(),
      unique (user_id, market_id, side)
    )
  `;

  await db`
    create table if not exists personal_capital_accounts (
      user_id text primary key,
      cash numeric not null default 0,
      updated_at timestamptz not null default now()
    )
  `;

  await db`
    create table if not exists personal_agent_allocations (
      user_id text not null,
      agent_id text not null,
      amount numeric not null default 0,
      updated_at timestamptz not null default now(),
      primary key (user_id, agent_id)
    )
  `;

  await db`
    create table if not exists personal_capital_events (
      id bigserial primary key,
      user_id text not null,
      action text not null,
      agent_id text,
      amount numeric not null,
      note text,
      created_at timestamptz not null default now()
    )
  `;

  await db`
    create index if not exists personal_capital_events_user_idx
      on personal_capital_events (user_id, created_at desc)
  `;

  schemaReady = true;
}

export async function recordProviderEvent(event) {
  await ensureSchema();
  const db = sql();
  const payload = event.payload && typeof event.payload === "object" ? event.payload : {};
  const headers = event.headers && typeof event.headers === "object" ? event.headers : {};

  const rows = await db`
    insert into provider_events (
      provider,
      event_type,
      external_id,
      signature_valid,
      payload,
      headers
    ) values (
      ${event.provider},
      ${event.eventType || "unknown"},
      ${event.externalId || null},
      ${Boolean(event.signatureValid)},
      ${JSON.stringify(payload)}::jsonb,
      ${JSON.stringify(headers)}::jsonb
    )
    on conflict do nothing
    returning id, received_at
  `;

  return rows[0] || null;
}

export async function recordAuditEvent(eventType, detail) {
  await ensureSchema();
  const db = sql();
  await db`
    insert into readiness_audit (event_type, detail)
    values (${eventType}, ${JSON.stringify(detail || {})}::jsonb)
  `;
}

export async function recordUserConsent(consent) {
  await ensureSchema();
  const db = sql();
  const rows = await db`
    insert into user_consents (
      user_id,
      terms_version,
      privacy_version,
      risk_disclosure_version,
      jurisdiction,
      payload
    ) values (
      ${consent.userId},
      ${consent.termsVersion},
      ${consent.privacyVersion},
      ${consent.riskDisclosureVersion},
      ${consent.jurisdiction || null},
      ${JSON.stringify(consent.payload || {})}::jsonb
    )
    returning id, accepted_at
  `;
  return rows[0] || null;
}

export async function upsertRiskProfile(profile) {
  await ensureSchema();
  const db = sql();
  const rows = await db`
    insert into risk_profiles (
      user_id,
      wallet_address,
      jurisdiction,
      deposit_limit,
      withdraw_reserve,
      agent_permissions,
      updated_at
    ) values (
      ${profile.userId},
      ${profile.walletAddress || null},
      ${profile.jurisdiction || null},
      ${Number(profile.depositLimit || 0)},
      ${Number(profile.withdrawReserve || 0)},
      ${JSON.stringify(profile.agentPermissions || {})}::jsonb,
      now()
    )
    on conflict (user_id) do update set
      wallet_address = excluded.wallet_address,
      jurisdiction = excluded.jurisdiction,
      deposit_limit = excluded.deposit_limit,
      withdraw_reserve = excluded.withdraw_reserve,
      agent_permissions = excluded.agent_permissions,
      updated_at = now()
    returning user_id, updated_at
  `;
  return rows[0] || null;
}

export async function latestRiskProfile(userId) {
  await ensureSchema();
  const db = sql();
  const rows = await db`
    select
      user_id,
      wallet_address,
      jurisdiction,
      deposit_limit,
      withdraw_reserve,
      agent_permissions,
      updated_at
    from risk_profiles
    where user_id = ${userId}
    limit 1
  `;
  return rows[0] || null;
}

export async function createTradeTicket(ticket) {
  await ensureSchema();
  const db = sql();
  const rows = await db`
    insert into trade_tickets (
      user_id,
      agent_id,
      market_id,
      question,
      market_url,
      side,
      max_amount,
      limit_price,
      rationale,
      status,
      instructions
    ) values (
      ${ticket.userId},
      ${ticket.agentId},
      ${ticket.marketId},
      ${ticket.question || null},
      ${ticket.marketUrl || null},
      ${ticket.side},
      ${Number(ticket.maxAmount || 0)},
      ${ticket.limitPrice == null ? null : Number(ticket.limitPrice)},
      ${ticket.rationale || null},
      ${ticket.status || "staged"},
      ${JSON.stringify(ticket.instructions || {})}::jsonb
    )
    returning *
  `;
  return rows[0] || null;
}

export async function listTradeTickets(userId, limit = 50) {
  await ensureSchema();
  const db = sql();
  return db`
    select *
    from trade_tickets
    where user_id = ${userId}
    order by created_at desc
    limit ${Math.max(1, Math.min(Number(limit) || 50, 100))}
  `;
}

export async function updateTradeTicketStatus(userId, ticketId, status) {
  await ensureSchema();
  const db = sql();
  const rows = await db`
    update trade_tickets
    set status = ${status}, updated_at = now()
    where user_id = ${userId} and id = ${Number(ticketId)}
    returning *
  `;
  return rows[0] || null;
}

export async function recordRealFill(fill) {
  await ensureSchema();
  const db = sql();
  const action = String(fill.action || "BUY").toUpperCase();
  const shares = Number(fill.shares || 0);
  const price = Number(fill.price || 0);
  const fees = Number(fill.fees || 0);

  const rows = await db`
    insert into real_fills (
      user_id,
      ticket_id,
      agent_id,
      market_id,
      question,
      market_url,
      side,
      action,
      shares,
      price,
      fees,
      tx_note,
      filled_at
    ) values (
      ${fill.userId},
      ${fill.ticketId ? Number(fill.ticketId) : null},
      ${fill.agentId || null},
      ${fill.marketId},
      ${fill.question || null},
      ${fill.marketUrl || null},
      ${fill.side},
      ${action},
      ${shares},
      ${price},
      ${fees},
      ${fill.txNote || null},
      ${fill.filledAt ? new Date(fill.filledAt).toISOString() : new Date().toISOString()}
    )
    returning *
  `;

  const existing = await db`
    select *
    from real_positions
    where user_id = ${fill.userId}
      and market_id = ${fill.marketId}
      and side = ${fill.side}
    limit 1
  `;
  const pos = existing[0];

  if (action === "SELL" && pos) {
    const sellShares = Math.min(shares, Number(pos.shares || 0));
    const avgPrice = Number(pos.avg_price || 0);
    const proceeds = sellShares * price - fees;
    const costRemoved = sellShares * avgPrice;
    const nextShares = Number(pos.shares || 0) - sellShares;
    const nextCost = Math.max(0, Number(pos.cost_basis || 0) - costRemoved);
    const realized = Number(pos.realized_pnl || 0) + proceeds - costRemoved;
    await db`
      update real_positions
      set shares = ${nextShares},
          cost_basis = ${nextCost},
          realized_pnl = ${realized},
          current_price = ${price},
          updated_at = now()
      where id = ${pos.id}
    `;
  } else {
    const currentShares = pos ? Number(pos.shares || 0) : 0;
    const currentCost = pos ? Number(pos.cost_basis || 0) : 0;
    const nextShares = currentShares + shares;
    const nextCost = currentCost + shares * price + fees;
    const avgPrice = nextShares > 0 ? nextCost / nextShares : 0;
    await db`
      insert into real_positions (
        user_id,
        agent_id,
        market_id,
        question,
        market_url,
        side,
        shares,
        avg_price,
        cost_basis,
        realized_pnl,
        current_price,
        updated_at
      ) values (
        ${fill.userId},
        ${fill.agentId || null},
        ${fill.marketId},
        ${fill.question || null},
        ${fill.marketUrl || null},
        ${fill.side},
        ${nextShares},
        ${avgPrice},
        ${nextCost},
        ${pos ? Number(pos.realized_pnl || 0) : 0},
        ${price},
        now()
      )
      on conflict (user_id, market_id, side) do update set
        agent_id = coalesce(excluded.agent_id, real_positions.agent_id),
        question = coalesce(excluded.question, real_positions.question),
        market_url = coalesce(excluded.market_url, real_positions.market_url),
        shares = excluded.shares,
        avg_price = excluded.avg_price,
        cost_basis = excluded.cost_basis,
        current_price = excluded.current_price,
        updated_at = now()
    `;
  }

  if (fill.ticketId) {
    await updateTradeTicketStatus(fill.userId, fill.ticketId, "placed_manually");
  }

  return rows[0] || null;
}

export async function updateRealPositionMark(userId, positionId, currentPrice) {
  await ensureSchema();
  const db = sql();
  const rows = await db`
    update real_positions
    set current_price = ${Number(currentPrice)}, updated_at = now()
    where user_id = ${userId} and id = ${Number(positionId)}
    returning *
  `;
  return rows[0] || null;
}

export async function listRealPortfolio(userId) {
  await ensureSchema();
  const db = sql();
  const positions = await db`
    select *
    from real_positions
    where user_id = ${userId}
    order by updated_at desc
  `;
  const fills = await db`
    select *
    from real_fills
    where user_id = ${userId}
    order by filled_at desc
    limit 100
  `;
  return { positions, fills };
}

export async function listPersonalCapital(userId) {
  await ensureSchema();
  const db = sql();
  await db`
    insert into personal_capital_accounts (user_id, cash)
    values (${userId}, 0)
    on conflict (user_id) do nothing
  `;
  const accounts = await db`
    select *
    from personal_capital_accounts
    where user_id = ${userId}
    limit 1
  `;
  const allocations = await db`
    select *
    from personal_agent_allocations
    where user_id = ${userId} and amount > 0
    order by updated_at desc
  `;
  const events = await db`
    select *
    from personal_capital_events
    where user_id = ${userId}
    order by created_at desc
    limit 100
  `;
  return { account: accounts[0] || { user_id: userId, cash: 0 }, allocations, events };
}

export async function recordPersonalCapitalAction(action) {
  await ensureSchema();
  const db = sql();
  const userId = action.userId;
  const type = String(action.action || "").toUpperCase();
  const agentId = action.agentId || null;
  const amount = Number(action.amount || 0);

  await db`
    insert into personal_capital_accounts (user_id, cash)
    values (${userId}, 0)
    on conflict (user_id) do nothing
  `;

  const accounts = await db`
    select *
    from personal_capital_accounts
    where user_id = ${userId}
    limit 1
  `;
  const account = accounts[0] || { cash: 0 };
  let nextCash = Number(account.cash || 0);

  if (type === "DEPOSIT") {
    nextCash += amount;
  } else if (type === "WITHDRAW") {
    nextCash -= amount;
  } else if (type === "BUY_AGENT") {
    nextCash -= amount;
    await db`
      insert into personal_agent_allocations (user_id, agent_id, amount, updated_at)
      values (${userId}, ${agentId}, ${amount}, now())
      on conflict (user_id, agent_id) do update set
        amount = personal_agent_allocations.amount + excluded.amount,
        updated_at = now()
    `;
  } else if (type === "SELL_AGENT") {
    const rows = await db`
      select *
      from personal_agent_allocations
      where user_id = ${userId} and agent_id = ${agentId}
      limit 1
    `;
    const current = rows[0] ? Number(rows[0].amount || 0) : 0;
    const reduction = Math.min(amount, current);
    nextCash += reduction;
    await db`
      insert into personal_agent_allocations (user_id, agent_id, amount, updated_at)
      values (${userId}, ${agentId}, ${Math.max(0, current - reduction)}, now())
      on conflict (user_id, agent_id) do update set
        amount = excluded.amount,
        updated_at = now()
    `;
  }

  if (nextCash < -0.000001) {
    throw new Error("Tracked personal cash is too low for that action");
  }

  await db`
    update personal_capital_accounts
    set cash = ${nextCash}, updated_at = now()
    where user_id = ${userId}
  `;

  await db`
    insert into personal_capital_events (user_id, action, agent_id, amount, note)
    values (${userId}, ${type}, ${agentId}, ${amount}, ${action.note || null})
  `;

  return listPersonalCapital(userId);
}

export async function providerEventSummary() {
  await ensureSchema();
  const db = sql();
  const rows = await db`
    select
      provider,
      count(*)::int as total,
      max(received_at) as latest_received_at,
      count(*) filter (where signature_valid)::int as signature_valid_count
    from provider_events
    group by provider
    order by provider
  `;
  return rows;
}